(kafka) branch 4.3 updated: KAFKA-20572 Fix NPE when using listener-prefixed SSL configs (#22269)

[chia7712]

This is an automated email from the ASF dual-hosted git repository.

chia7712 pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/4.3 by this push:
     new 25ecb749f5e KAFKA-20572 Fix NPE when using listener-prefixed SSL 
configs (#22269)
25ecb749f5e is described below

commit 25ecb749f5e12a429f878bb1c99913d5a6426147
Author: Hector Geraldino <hgerald...@bloomberg.net>
AuthorDate: Tue May 12 22:32:37 2026 -0400

    KAFKA-20572 Fix NPE when using listener-prefixed SSL configs (#22269)
    
    Fixes NPE when starting Kafka Connect 4.2.0 with listener-prefixed SSL
    configs.
    
    JIRA: https://issues.apache.org/jira/browse/KAFKA-20572
    
    This is a regression introduced on
    https://github.com/apache/kafka/pull/20334
    
    Reviewers: Chia-Ping Tsai <chia7...@gmail.com>
---
 .../kafka/connect/runtime/rest/util/SSLUtils.java  |  2 +-
 .../connect/runtime/rest/util/SSLUtilsTest.java    | 31 ++++++++++++++++++++++
 2 files changed, 32 insertions(+), 1 deletion(-)

diff --git 
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/util/SSLUtils.java
 
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/util/SSLUtils.java
index 83a175e8d5f..a8863f19be7 100644
--- 
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/util/SSLUtils.java
+++ 
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/util/SSLUtils.java
@@ -135,7 +135,7 @@ public class SSLUtils {
 
         List<String> sslCipherSuites = (List<String>) 
sslConfigValues.get(SslConfigs.SSL_CIPHER_SUITES_CONFIG);
 
-        if (!sslCipherSuites.isEmpty())
+        if (sslCipherSuites != null && !sslCipherSuites.isEmpty())
             ssl.setIncludeCipherSuites(sslCipherSuites.toArray(new String[0]));
 
         ssl.setKeyManagerFactoryAlgorithm((String) 
getOrDefault(sslConfigValues, SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG, 
SslConfigs.DEFAULT_SSL_KEYMANGER_ALGORITHM));
diff --git 
a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/util/SSLUtilsTest.java
 
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/util/SSLUtilsTest.java
index c9eedccdc64..d88ca5bda8a 100644
--- 
a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/util/SSLUtilsTest.java
+++ 
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/util/SSLUtilsTest.java
@@ -30,6 +30,7 @@ import java.util.List;
 import java.util.Map;
 
 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 static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -163,6 +164,36 @@ public class SSLUtilsTest {
         assertFalse(ssl.getWantClientAuth());
     }
 
+    @Test
+    public void 
testCreateServerSideSslContextFactoryWithListenerPrefixedConfigs() {
+        Map<String, String> configMap = new HashMap<>();
+        // Use listener-prefixed SSL configs (but not all of them).
+        // This triggers the "all" branch of valuesWithPrefixAllOrNothing,
+        // which returns only prefixed entries — keys without a prefixed
+        // counterpart (like ssl.cipher.suites) will be absent from the map.
+        configMap.put("listeners.https.ssl.keystore.location", keystorePath);
+        configMap.put("listeners.https.ssl.keystore.password", 
keystorePassword.value());
+        configMap.put("listeners.https.ssl.key.password", 
keystorePassword.value());
+        configMap.put("listeners.https.ssl.truststore.location", 
truststorePath);
+        configMap.put("listeners.https.ssl.truststore.password", 
truststorePassword.value());
+
+        RestServerConfig config = RestServerConfig.forPublic(null, configMap);
+        assertDoesNotThrow(() -> 
SSLUtils.createServerSideSslContextFactory(config));
+    }
+
+    @Test
+    public void 
testCreateClientSideSslContextFactoryWithListenerPrefixedConfigs() {
+        Map<String, String> configMap = new HashMap<>();
+        configMap.put("listeners.https.ssl.keystore.location", keystorePath);
+        configMap.put("listeners.https.ssl.keystore.password", 
keystorePassword.value());
+        configMap.put("listeners.https.ssl.key.password", 
keystorePassword.value());
+        configMap.put("listeners.https.ssl.truststore.location", 
truststorePath);
+        configMap.put("listeners.https.ssl.truststore.password", 
truststorePassword.value());
+
+        RestServerConfig config = RestServerConfig.forPublic(null, configMap);
+        assertDoesNotThrow(() -> 
SSLUtils.createClientSideSslContextFactory(config));
+    }
+
     @Test
     public void testCreateClientSideSslContextFactoryDefaultValues() {
         Map<String, String> configMap = new HashMap<>();