removed kafka.test.version definition and added consistent implementation of enqueueNewPartitions method

Author: ramitg254

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>3.9.1</kafka.test.version>
     <druid.guice.version>4.1.0</druid.guice.version>
     <slf4j.version>1.7.30</slf4j.version>
   </properties>
@@ -219,17 +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.test.version}</version>
+      <version>${kafka.version}</version>
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>

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

@@ -45,7 +45,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.Set;
 import java.util.concurrent.Future;
 
 /**
@@ -188,27 +187,43 @@ short getEpoch() {
    */
   private void flushNewPartitions() {
     LOG.info("Flushing new partitions");
-    Object transactionManager = getValue(kafkaProducer, "transactionManager");
-    Set<TopicPartition> newPartitionsInTransaction =
-            (Set<TopicPartition>) getValue(transactionManager, "newPartitionsInTransaction");
-    if (!newPartitionsInTransaction.isEmpty()) {
-      TransactionalRequestResult result = enqueueNewPartitions();
-      Object sender = getValue(kafkaProducer, "sender");
-      invoke(sender, "wakeup");
-      result.await();
-    }
+    TransactionalRequestResult result = enqueueNewPartitions();
+    Object sender = getValue(kafkaProducer, "sender");
+    invoke(sender, "wakeup");
+    result.await();
   }
 
   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) {