(camel) 01/01: CAMEL-20549: camel-kafka - Using sslKeystoreType should work with PEM

[davsclaus]

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

davsclaus pushed a commit to branch pem
in repository https://gitbox.apache.org/repos/asf/camel.git

commit 9b4d2d576e2449d25b6e7bf5825b3fb94cdf04ac
Author: Claus Ibsen <claus.ib...@gmail.com>
AuthorDate: Mon Mar 11 14:11:13 2024 +0100

    CAMEL-20549: camel-kafka - Using sslKeystoreType should work with PEM
---
 .../camel/component/kafka/KafkaConfiguration.java       | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

diff --git 
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
 
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
index 58ab3220db7..0e9b05e9f50 100755
--- 
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
+++ 
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.kafka;
 
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.ExecutorService;
@@ -441,10 +442,10 @@ public class KafkaConfiguration implements Cloneable, 
HeaderFilterStrategyAware
             addPropertyIfNotEmpty(props, 
SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, getSslTruststoreLocation());
             addPropertyIfNotEmpty(props, 
SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, getSslTruststorePassword());
             addPropertyIfNotEmpty(props, 
SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, getSslEnabledProtocols());
-            addPropertyIfNotEmpty(props, SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, 
getSslKeystoreType());
+            addUpperCasePropertyIfNotEmpty(props, 
SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, getSslKeystoreType());
             addPropertyIfNotEmpty(props, SslConfigs.SSL_PROTOCOL_CONFIG, 
getSslProtocol());
             addPropertyIfNotEmpty(props, SslConfigs.SSL_PROVIDER_CONFIG, 
getSslProvider());
-            addPropertyIfNotEmpty(props, 
SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, getSslTruststoreType());
+            addUpperCasePropertyIfNotEmpty(props, 
SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, getSslTruststoreType());
             addPropertyIfNotEmpty(props, SslConfigs.SSL_CIPHER_SUITES_CONFIG, 
getSslCipherSuites());
             String algo = getSslEndpointAlgorithm();
             if (algo != null && !algo.equals("none") && !algo.equals("false")) 
{
@@ -529,10 +530,10 @@ public class KafkaConfiguration implements Cloneable, 
HeaderFilterStrategyAware
             addPropertyIfNotEmpty(props, 
SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG, getSslKeymanagerAlgorithm());
             addPropertyIfNotEmpty(props, 
SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG, getSslTrustmanagerAlgorithm());
             addPropertyIfNotEmpty(props, 
SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, getSslEnabledProtocols());
-            addPropertyIfNotEmpty(props, SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, 
getSslKeystoreType());
+            addUpperCasePropertyIfNotEmpty(props, 
SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, getSslKeystoreType());
             addPropertyIfNotEmpty(props, SslConfigs.SSL_PROTOCOL_CONFIG, 
getSslProtocol());
             addPropertyIfNotEmpty(props, SslConfigs.SSL_PROVIDER_CONFIG, 
getSslProvider());
-            addPropertyIfNotEmpty(props, 
SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, getSslTruststoreType());
+            addUpperCasePropertyIfNotEmpty(props, 
SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, getSslTruststoreType());
             addPropertyIfNotEmpty(props, ProducerConfig.SEND_BUFFER_CONFIG, 
getSendBufferBytes());
         }
     }
@@ -564,7 +565,7 @@ public class KafkaConfiguration implements Cloneable, 
HeaderFilterStrategyAware
             addPropertyIfNotNull(props, SslConfigs.SSL_KEY_PASSWORD_CONFIG, 
keyManagers.getKeyPassword());
             KeyStoreParameters keyStore = keyManagers.getKeyStore();
             if (keyStore != null) {
-                addPropertyIfNotNull(props, 
SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, keyStore.getType());
+                addUpperCasePropertyIfNotEmpty(props, 
SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, keyStore.getType());
                 addPropertyIfNotNull(props, 
SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keyStore.getResource());
                 addPropertyIfNotNull(props, 
SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, keyStore.getPassword());
             }
@@ -607,6 +608,12 @@ public class KafkaConfiguration implements Cloneable, 
HeaderFilterStrategyAware
         }
     }
 
+    private static <T> void addUpperCasePropertyIfNotEmpty(Properties props, 
String key, T value) {
+        if (ObjectHelper.isNotEmpty(value)) {
+            props.put(key, String.valueOf(value).toUpperCase(Locale.ROOT));
+        }
+    }
+
     private static <T> void addPropertyIfNotNull(Properties props, String key, 
T value) {
         if (value != null) {
             // value should be as-is