fix(kafka): upgrade 3.9 client for SASL authentication on newer JDKs

Upgrade to Kafka 3.9.1, which includes the KAFKA-17078 Subject compatibility shim. Cover password-based SASL callbacks through that shim.

Closes #9589

agents/drivers/kafka/build.gradle

@@ -1,6 +1,6 @@
 dependencies {
     implementation 'com.google.code.gson:gson:2.12.1'
-    implementation 'org.apache.kafka:kafka-clients:3.9.0'
+    implementation 'org.apache.kafka:kafka-clients:3.9.1'
     implementation('org.apache.zookeeper:zookeeper:3.8.4') {
         exclude group: 'log4j', module: 'log4j'
         exclude group: 'org.slf4j', module: 'slf4j-log4j12'

agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaClientCompatibilityTest.java

@@ -0,0 +1,40 @@
+package com.dbx.agent.kafka;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.List;
+import java.util.Map;
+import javax.security.auth.Subject;
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import org.apache.kafka.common.internals.SecurityManagerCompatibility;
+import org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler;
+import org.junit.jupiter.api.Test;
+
+class KafkaClientCompatibilityTest {
+    @Test
+    void passwordSaslCallbacksUseTheCurrentSubject() {
+        // KAFKA-17078 routes Subject access through this shim when newer JDKs
+        // disable the legacy Subject.getSubject API.
+        Subject subject = new Subject();
+        subject.getPublicCredentials().add("dbx-user");
+        subject.getPrivateCredentials().add("dbx-password");
+
+        for (String mechanism : List.of("PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512")) {
+            SaslClientCallbackHandler handler = new SaslClientCallbackHandler();
+            handler.configure(Map.of(), mechanism, List.of());
+            NameCallback name = new NameCallback("username");
+            PasswordCallback password = new PasswordCallback("password", false);
+
+            assertDoesNotThrow(() -> SecurityManagerCompatibility.get().callAs(subject, () -> {
+                handler.handle(new Callback[] { name, password });
+                return null;
+            }));
+            assertEquals("dbx-user", name.getName());
+            assertArrayEquals("dbx-password".toCharArray(), password.getPassword());
+        }
+    }
+}