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

itests/qtest-druid/pom.xml

@@ -36,7 +36,6 @@
     <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>
     <druid.guice.version>4.1.0</druid.guice.version>
     <slf4j.version>1.7.30</slf4j.version>
   </properties>
@@ -219,12 +218,17 @@
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka_2.12</artifactId>
-      <version>${kafka.test.version}</version>
+      <version>${kafka.version}</version>
     </dependency>
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-clients</artifactId>
-      <version>${kafka.test.version}</version>
+      <version>${kafka.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-server</artifactId>
+      <version>${kafka.version}</version>
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>

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;
 
@@ -81,8 +83,8 @@ public SingleNodeKafkaCluster(String name, String logDir, Integer zkPort, Intege
 
     properties.setProperty("zookeeper.connect", zkString);
     properties.setProperty("broker.id", String.valueOf(1));
-    properties.setProperty("host.name", LOCALHOST);
-    properties.setProperty("port", Integer.toString(brokerPort));
+    properties.setProperty("listeners", "PLAINTEXT://" + LOCALHOST + ":" + Integer.toString(brokerPort));
+    properties.setProperty("advertised.listeners", "PLAINTEXT://" + LOCALHOST + ":" + Integer.toString(brokerPort));
     properties.setProperty("log.dir", logDir);
     // This property is very important, we are sending form records with a specific time
     // Thus need to make sure that they don't get DELETED
@@ -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();
   }
@@ -138,11 +144,11 @@ synchronized void resumeTransaction(long producerId, short epoch) {
 
     Object transactionManager = getValue(kafkaProducer, "transactionManager");
 
-    Object topicPartitionBookkeeper = getValue(transactionManager, "topicPartitionBookkeeper");
+    Object txnPartitionMap = getValue(transactionManager, "txnPartitionMap");
     invoke(transactionManager,
         "transitionTo",
         getEnum("org.apache.kafka.clients.producer.internals.TransactionManager$State.INITIALIZING"));
-    invoke(topicPartitionBookkeeper, "reset");
+    invoke(txnPartitionMap, "reset");
     Object producerIdAndEpoch = getValue(transactionManager, "producerIdAndEpoch");
     setValue(producerIdAndEpoch, "producerId", producerId);
     setValue(producerIdAndEpoch, "epoch", epoch);
@@ -189,14 +195,35 @@ private void flushNewPartitions() {
 
   private synchronized TransactionalRequestResult enqueueNewPartitions() {
     Object transactionManager = getValue(kafkaProducer, "transactionManager");
-    Object txnRequestHandler = invoke(transactionManager, "addPartitionsToTransactionHandler");
-    invoke(transactionManager,
-        "enqueueRequest",
-        new Class[] {txnRequestHandler.getClass().getSuperclass()},
-        new Object[] {txnRequestHandler});
-    return (TransactionalRequestResult) getValue(txnRequestHandler,
-        txnRequestHandler.getClass().getSuperclass(),
-        "result");
+    synchronized (transactionManager) {
+      Object newPartitionsInTransaction =
+              getValue(transactionManager, "newPartitionsInTransaction");
+      Object newPartitionsInTransactionIsEmpty =
+              invoke(newPartitionsInTransaction, "isEmpty");
+      TransactionalRequestResult result;
+      if (newPartitionsInTransactionIsEmpty instanceof Boolean
+              && !((Boolean) newPartitionsInTransactionIsEmpty)) {
+        Object txnRequestHandler =
+                invoke(transactionManager, "addPartitionsToTransactionHandler");
+        invoke(
+                transactionManager,
+                "enqueueRequest",
+                new Class[]{txnRequestHandler.getClass().getSuperclass()},
+                new Object[]{txnRequestHandler});
+
+        result = (TransactionalRequestResult)
+                getValue(
+                        txnRequestHandler,
+                        txnRequestHandler.getClass().getSuperclass(),
+                        "result");
+      } else {
+        // we don't have an operation but this operation string is also used in
+        // addPartitionsToTransactionHandler.
+        result = new TransactionalRequestResult("AddPartitionsToTxn");
+        result.done();
+      }
+      return result;
+    }
   }
 
   @SuppressWarnings("unchecked") private static Enum<?> getEnum(String enumFullName) {

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

@@ -140,7 +140,7 @@ class KafkaRecordIterator implements Iterator<ConsumerRecord<byte[], byte[]>> {
       }
     } else {
       // case seek to beginning of stream
-      consumer.seekToBeginning(Collections.singleton(topicPartition));
+      consumer.seekToBeginning(topicPartitionList);
       // seekToBeginning is lazy thus need to call position() or poll(0)
       this.startOffset = consumer.position(topicPartition);
       LOG.info("Consumer at beginning of topic partition [{}], current start offset [{}]",

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

@@ -30,6 +30,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
 import java.util.Iterator;
 import java.util.Properties;
 
@@ -150,7 +151,7 @@ private synchronized void initialize(KafkaInputSplit inputSplit, Configuration j
     LOG.trace("total read bytes [{}]", readBytes);
     if (consumer != null) {
       consumer.wakeup();
-      consumer.close();
+      consumer.close(Duration.ZERO);
     }
   }
 

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

@@ -37,6 +37,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.Iterator;
 import java.util.Properties;
 
@@ -150,7 +151,7 @@ private void cleanRowBoat() {
     LOG.trace("total read bytes [{}]", readBytes);
     if (consumer != null) {
       consumer.wakeup();
-      consumer.close();
+      consumer.close(Duration.ZERO);
     }
   }
 

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

@@ -25,6 +25,7 @@
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
@@ -158,7 +159,9 @@
   @Test(expected = org.apache.kafka.common.KafkaException.class) public void testWrongEpochAndId() {
     HiveKafkaProducer secondProducer = new HiveKafkaProducer(producerProperties);
     secondProducer.resumeTransaction(3434L, (short) 12);
-    secondProducer.sendOffsetsToTransaction(ImmutableMap.of(), "__dummy_consumer_group");
+    secondProducer.sendOffsetsToTransaction(Collections.singletonMap(
+            new TopicPartition("dummy_topic", 0),
+            new OffsetAndMetadata(0L)), "__dummy_consumer_group");
     secondProducer.close(Duration.ZERO);
   }
 
@@ -169,7 +172,9 @@
     producer.close(Duration.ZERO);
     HiveKafkaProducer secondProducer = new HiveKafkaProducer(producerProperties);
     secondProducer.resumeTransaction(pid, (short) 12);
-    secondProducer.sendOffsetsToTransaction(ImmutableMap.of(), "__dummy_consumer_group");
+    secondProducer.sendOffsetsToTransaction(Collections.singletonMap(
+            new TopicPartition("dummy_topic", 0),
+            new OffsetAndMetadata(0L)), "__dummy_consumer_group");
     secondProducer.close(Duration.ZERO);
   }
 
@@ -180,7 +185,9 @@
     producer.close(Duration.ZERO);
     HiveKafkaProducer secondProducer = new HiveKafkaProducer(producerProperties);
     secondProducer.resumeTransaction(45L, epoch);
-    secondProducer.sendOffsetsToTransaction(ImmutableMap.of(), "__dummy_consumer_group");
+    secondProducer.sendOffsetsToTransaction(Collections.singletonMap(
+            new TopicPartition("dummy_topic", 0),
+            new OffsetAndMetadata(0L)), "__dummy_consumer_group");
     secondProducer.close(Duration.ZERO);
   }
 }

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;
@@ -41,6 +41,7 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Collectors;
+import scala.Option;
 
 /**
  * Test Helper Class to start and stop a kafka broker.
@@ -106,7 +107,8 @@ 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 +118,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(), Option.empty());
     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);
   }
 
   /**

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

@@ -44,6 +44,7 @@
 
 import javax.annotation.Nullable;
 import java.nio.charset.Charset;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
@@ -304,7 +305,7 @@ private static void sendData(List<ConsumerRecord<byte[], byte[]>> recordList, @N
   @After public void tearDown() {
     this.kafkaRecordIterator = null;
     if (this.consumer != null) {
-      this.consumer.close();
+      this.consumer.close(Duration.ZERO);
     }
   }
 

pom.xml

@@ -173,7 +173,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>

ql/src/test/queries/clientpositive/kafka_storage_handler.q

@@ -1,5 +1,3 @@
---! qt:disabled:HIVE-23985
-
 SET hive.vectorized.execution.enabled=true;
 
 CREATE EXTERNAL TABLE kafka_table