HIVE-29238:upgrade kafka version to fix CVE-2024-31141 and CVE-2021-38153

Author: ramitg254

itests/qtest-druid/pom.xml

@@ -36,7 +36,7 @@
     <druid.derby.version>10.11.1.1</druid.derby.version>
     <druid.guava.version>16.0.1</druid.guava.version>
     <druid.guice.version>4.1.0</druid.guice.version>
-    <kafka.test.version>2.5.0</kafka.test.version>
+    <kafka.test.version>3.9.1</kafka.test.version>
     <druid.guice.version>4.1.0</druid.guice.version>
     <slf4j.version>1.7.30</slf4j.version>
   </properties>
@@ -226,6 +226,11 @@
       <artifactId>kafka-clients</artifactId>
       <version>${kafka.test.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-server</artifactId>
+      <version>${kafka.test.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>

itests/qtest-druid/src/main/java/org/apache/hive/kafka/SingleNodeKafkaCluster.java

@@ -19,7 +19,7 @@
 package org.apache.hive.kafka;
 
 import kafka.server.KafkaConfig;
-import kafka.server.KafkaServerStartable;
+import kafka.server.KafkaServer;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.service.AbstractService;
@@ -29,6 +29,7 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.utils.Time;
 
 import com.google.common.base.Throwables;
 import com.google.common.io.Files;
@@ -43,6 +44,7 @@
 import java.util.List;
 import java.util.Properties;
 import java.util.stream.IntStream;
+import scala.Option;
 
 /**
  * This class has the hooks to start and stop single node kafka cluster.
@@ -54,7 +56,7 @@ public class SingleNodeKafkaCluster extends AbstractService {
   private static final String LOCALHOST = "localhost";
 
 
-  private final KafkaServerStartable serverStartable;
+  private final KafkaServer server;
   private final int brokerPort;
   private final String kafkaServer;
 
@@ -94,21 +96,21 @@ public SingleNodeKafkaCluster(String name, String logDir, Integer zkPort, Intege
     properties.setProperty("transaction.state.log.min.isr", String.valueOf(1));
     properties.setProperty("log.cleaner.dedupe.buffer.size", "1048577");
 
-    this.serverStartable = new KafkaServerStartable(KafkaConfig.fromProps(properties));
+    this.server = new KafkaServer(KafkaConfig.fromProps(properties),Time.SYSTEM, Option.empty(), false);
   }
 
 
   @Override
   protected void serviceStart() throws Exception {
-    serverStartable.startup();
+    server.startup();
     log.info("Kafka Server Started on port {}", brokerPort);
 
   }
 
   @Override
   protected void serviceStop() throws Exception {
     log.info("Stopping Kafka Server");
-    serverStartable.shutdown();
+    server.shutdown();
     log.info("Kafka Server Stopped");
   }
 

kafka-handler/src/java/org/apache/hadoop/hive/kafka/HiveKafkaProducer.java

@@ -33,6 +33,7 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.Uuid;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -67,6 +68,11 @@ class HiveKafkaProducer<K, V> implements Producer<K, V> {
     kafkaProducer = new KafkaProducer<>(properties);
   }
 
+  @Override
+  public Uuid clientInstanceId(Duration timeout) {
+    throw new UnsupportedOperationException();
+  }
+
   @Override public void initTransactions() {
     kafkaProducer.initTransactions();
   }

kafka-handler/src/test/org/apache/hadoop/hive/kafka/KafkaBrokerResource.java

@@ -26,7 +26,7 @@
 import kafka.zk.EmbeddedZookeeper;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.hive.common.IPStackUtils;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.test.TestSslUtils;
 import org.junit.rules.ExternalResource;
@@ -106,7 +106,7 @@ KafkaBrokerResource enableSASL(String principal, String keytab) {
       brokerProps.setProperty("listener.name.l2.gssapi.sasl.jaas.config", jaasConfig);
       brokerProps.setProperty("listener.name.l3.gssapi.sasl.jaas.config", jaasConfig);
       truststoreFile = File.createTempFile("kafka_truststore", "jks");
-      brokerProps.putAll(new TestSslUtils.SslConfigsBuilder(Mode.SERVER).createNewTrustStore(truststoreFile).build());
+      brokerProps.putAll(new TestSslUtils.SslConfigsBuilder(ConnectionMode.SERVER).createNewTrustStore(truststoreFile).build());
       brokerProps.setProperty("delegation.token.master.key", "AnyValueShouldDoHereItDoesntMatter");
     }
     brokerProps.setProperty("offsets.topic.replication.factor", "1");
@@ -116,9 +116,9 @@ KafkaBrokerResource enableSASL(String principal, String keytab) {
     kafkaServer = TestUtils.createServer(config, Time.SYSTEM);
     kafkaServer.startup();
     kafkaServer.zkClient();
-    adminZkClient = new AdminZkClient(kafkaServer.zkClient());
+    adminZkClient = new AdminZkClient(kafkaServer.zkClient(),null);
     LOG.info("Creating kafka TOPIC [{}]", TOPIC);
-    adminZkClient.createTopic(TOPIC, 1, 1, new Properties(), RackAwareMode.Disabled$.MODULE$);
+    adminZkClient.createTopic(TOPIC, 1, 1, new Properties(), RackAwareMode.Disabled$.MODULE$,false);
   }
 
   /**

pom.xml

@@ -172,7 +172,7 @@
     <junit.version>4.13.2</junit.version>
     <junit.jupiter.version>5.13.3</junit.jupiter.version>
     <junit.vintage.version>5.13.3</junit.vintage.version>
-    <kafka.version>2.5.0</kafka.version>
+    <kafka.version>3.9.1</kafka.version>
     <kryo.version>5.5.0</kryo.version>
     <reflectasm.version>1.11.9</reflectasm.version>
     <kudu.version>1.17.0</kudu.version>