From 6355f902aeb34a0d104ad22dee2afe73bf945e96 Mon Sep 17 00:00:00 2001
From: Ted Yu <yuzhihong@gmail.com>
Date: Tue, 22 Nov 2022 02:44:43 -0800
Subject: [PATCH 1/3] [SPARK-41197] Upgrade Kafka version to 3.3 release

---
 .../scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala    | 2 ++
 .../org/apache/spark/streaming/kafka010/KafkaTestUtils.scala    | 2 ++
 pom.xml                                                         | 2 +-
 3 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
index 431d9d6b27879..b7c0fa27273da 100644
--- a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
+++ b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
@@ -537,6 +537,8 @@ class KafkaTestUtils(
     props.put("key.serializer", classOf[StringSerializer].getName)
     // wait for all in-sync replicas to ack sends
     props.put("acks", "all")
+    props.put("partitioner.class",
+        classOf[org.apache.kafka.clients.producer.internals.DefaultPartitioner].getName)
     setAuthenticationConfigIfNeeded(props)
     props
   }
diff --git a/connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala b/connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala
index d341b6977b25e..931eda6cb2b7f 100644
--- a/connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala
+++ b/connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala
@@ -263,6 +263,8 @@ private[kafka010] class KafkaTestUtils extends Logging {
     props.put("key.serializer", classOf[StringSerializer].getName)
     // wait for all in-sync replicas to ack sends
     props.put("acks", "all")
+    props.put("partitioner.class",
+        classOf[org.apache.kafka.clients.producer.internals.DefaultPartitioner].getName)
     props
   }
 
diff --git a/pom.xml b/pom.xml
index 2dd898b8787e0..932899106cc50 100644
--- a/pom.xml
+++ b/pom.xml
@@ -130,7 +130,7 @@
     <!-- Version used for internal directory structure -->
     <hive.version.short>2.3</hive.version.short>
     <!-- note that this should be compatible with Kafka brokers version 0.10 and up -->
-    <kafka.version>3.2.3</kafka.version>
+    <kafka.version>3.3.1</kafka.version>
     <!-- After 10.15.1.3, the minimum required version is JDK9 -->
     <derby.version>10.14.2.0</derby.version>
     <parquet.version>1.12.3</parquet.version>

From 264d2a99784c92b31904cbbea514ae3905b83a87 Mon Sep 17 00:00:00 2001
From: Dongjoon Hyun <dongjoon@apache.org>
Date: Mon, 28 Nov 2022 12:48:14 -0800
Subject: [PATCH 2/3] Fix indentation

---
 .../scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala    | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
index b7c0fa27273da..7c9c40883a58f 100644
--- a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
+++ b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
@@ -538,7 +538,7 @@ class KafkaTestUtils(
     // wait for all in-sync replicas to ack sends
     props.put("acks", "all")
     props.put("partitioner.class",
-        classOf[org.apache.kafka.clients.producer.internals.DefaultPartitioner].getName)
+      classOf[org.apache.kafka.clients.producer.internals.DefaultPartitioner].getName)
     setAuthenticationConfigIfNeeded(props)
     props
   }

From de32457d9a5bbf0b8efc640a9893dff60a7a0aa6 Mon Sep 17 00:00:00 2001
From: Dongjoon Hyun <dongjoon@apache.org>
Date: Mon, 28 Nov 2022 12:48:55 -0800
Subject: [PATCH 3/3] Fix indentation

---
 .../org/apache/spark/streaming/kafka010/KafkaTestUtils.scala    | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala b/connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala
index 931eda6cb2b7f..91fecacb6e724 100644
--- a/connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala
+++ b/connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala
@@ -264,7 +264,7 @@ private[kafka010] class KafkaTestUtils extends Logging {
     // wait for all in-sync replicas to ack sends
     props.put("acks", "all")
     props.put("partitioner.class",
-        classOf[org.apache.kafka.clients.producer.internals.DefaultPartitioner].getName)
+      classOf[org.apache.kafka.clients.producer.internals.DefaultPartitioner].getName)
     props
   }