Update pinned kafka version to 3.9.0 to support OAuth in python pipelines on KafkaIO. (#34197)

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json

 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 1
+  "modification": 2
 }

sdks/java/io/expansion-service/build.gradle

@@ -27,10 +27,12 @@ applyJavaNature(
   shadowClosure: {},
 )
 
-// TODO(https://github.com/apache/beam/pull/32486/) Use library.java.kafka_clients once >=3.1.0 is set as default
+// We don't want to use the latest version for the entire beam sdk since beam Java users can override it themselves.
+// We set this for python and cross-lang since it cannot be overwritten and version 3.4.0>= is required in order to
+// take advantage of OAuth authentication with all providers.
 configurations.runtimeClasspath {
-  // Pin kafka-clients version due to <3.1.0 missing auth callback classes
-  resolutionStrategy.force 'org.apache.kafka:kafka-clients:3.1.2'
+  // Pin kafka-clients version due to <3.4.0 missing auth callback classes.
+  resolutionStrategy.force 'org.apache.kafka:kafka-clients:3.9.0'
 }
 
 shadowJar {

sdks/java/io/kafka/build.gradle

@@ -43,6 +43,7 @@ def kafkaVersions = [
     '241': "2.4.1",
     '251': "2.5.1",
     '312': "3.1.2",
+    '390': "3.9.0",
 ]
 
 kafkaVersions.each{k,v -> configurations.create("kafkaVersion$k")}
@@ -72,7 +73,7 @@ dependencies {
   implementation library.java.jackson_annotations
   implementation library.java.jackson_databind
   implementation "org.springframework:spring-expression:5.3.27"
-  implementation group: 'com.google.cloud.hosted.kafka', name: 'managed-kafka-auth-login-handler', version: '1.0.2'
+  implementation group: 'com.google.cloud.hosted.kafka', name: 'managed-kafka-auth-login-handler', version: '1.0.5'
   implementation ("io.confluent:kafka-avro-serializer:${confluentVersion}") {
     // zookeeper depends on "spotbugs-annotations:3.1.9" which clashes with current
     // "spotbugs-annotations:3.1.12" used in Beam. Not required.
@@ -145,6 +146,7 @@ task kafkaVersionsCompatibilityTest {
   dependsOn (":sdks:java:io:kafka:kafka-241:kafkaVersion241BatchIT")
   dependsOn (":sdks:java:io:kafka:kafka-251:kafkaVersion251BatchIT")
   dependsOn (":sdks:java:io:kafka:kafka-312:kafkaVersion312BatchIT")
+  dependsOn (":sdks:java:io:kafka:kafka-390:kafkaVersion390BatchIT")
 }
 
 static def createTestList(Map<String, String> prefixMap, String suffix) {

sdks/java/io/kafka/kafka-390/build.gradle

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+project.ext {
+    delimited="3.9.0"
+    undelimited="390"
+    sdfCompatible=true
+}
+
+apply from: "../kafka-integration-test.gradle"
\ No newline at end of file

settings.gradle.kts

@@ -335,6 +335,8 @@ project(":beam-test-gha").projectDir = file(".github")
 include("beam-validate-runner")
 project(":beam-validate-runner").projectDir = file(".test-infra/validate-runner")
 include("com.google.api.gax.batching")
+include("sdks:java:io:kafka:kafka-390")
+findProject(":sdks:java:io:kafka:kafka-390")?.name = "kafka-390"
 include("sdks:java:io:kafka:kafka-312")
 findProject(":sdks:java:io:kafka:kafka-312")?.name = "kafka-312"
 include("sdks:java:io:kafka:kafka-251")