fix: Update kafka-clients to 3.8.0 past snappy CVEs and implement dum…

…my `clientInstanceId` (#482)
googleapis · Oct 2, 2024 · 06062a4 · 06062a4
1 parent 63a590e
commit 06062a4
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 1 deletion.
diff --git a/pom.xml b/pom.xml
@@ -36,7 +36,7 @@
       <dependency>
         <groupId>org.apache.kafka</groupId>
         <artifactId>kafka-clients</artifactId>
-        <version>3.4.0</version>
+        <version>3.8.0</version>
       </dependency>
     </dependencies>
   </dependencyManagement>

diff --git a/pubsublite-kafka/src/main/java/com/google/cloud/pubsublite/kafka/PubsubLiteConsumer.java b/pubsublite-kafka/src/main/java/com/google/cloud/pubsublite/kafka/PubsubLiteConsumer.java
@@ -64,6 +64,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 
 /**
  * A class that uses a SingleSubscriptionConsumer to remove the duplicate methods from the kafka
@@ -121,6 +122,12 @@ private SingleSubscriptionConsumer requireValidConsumer() {
     return consumer.get();
   }
 
+  @Override
+  public Uuid clientInstanceId(Duration timeout) {
+    // https://javadoc.io/static/org.apache.kafka/kafka-clients/3.8.0/org/apache/kafka/clients/consumer/KafkaConsumer.html#clientInstanceId-java.time.Duration-
+    throw new IllegalStateException("Pub/Sub Lite Kafka Connector does not support telemetry");
+  }
+
   @Override
   public Set<TopicPartition> assignment() {
     return requireValidConsumer().assignment().stream()

diff --git a/pubsublite-kafka/src/main/java/com/google/cloud/pubsublite/kafka/PubsubLiteProducer.java b/pubsublite-kafka/src/main/java/com/google/cloud/pubsublite/kafka/PubsubLiteProducer.java
@@ -48,6 +48,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 
 class PubsubLiteProducer implements Producer<byte[], byte[]> {
@@ -77,6 +78,12 @@ public void failed(State from, Throwable failure) {
     this.publisher.startAsync().awaitRunning();
   }
 
+  @Override
+  public Uuid clientInstanceId(Duration timeout) {
+    // https://javadoc.io/static/org.apache.kafka/kafka-clients/3.8.0/org/apache/kafka/clients/consumer/KafkaConsumer.html#clientInstanceId-java.time.Duration-
+    throw new IllegalStateException("Pub/Sub Lite Kafka Connector does not support telemetry");
+  }
+
   @Override
   public void initTransactions() {
     throw NO_TRANSACTIONS_EXCEPTION;