This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 113088da6cda CAMEL-23995: Fix saslAuthType overriding explicit
securityProtocol and generating non-working configs
113088da6cda is described below
commit 113088da6cdad653f12eec85e151e460aafeb01e
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 09:02:41 2026 +0200
CAMEL-23995: Fix saslAuthType overriding explicit securityProtocol and
generating non-working configs
Fix three bugs in the saslAuthType convenience layer (CAMEL-22864) that made
3 of 5 SASL auth types non-functional and silently overrode explicit user
configuration:
- SSL tautology: saslAuthType != NONE was always true inside isSasl(),
forcing
SASL_SSL even when user set securityProtocol=SASL_PLAINTEXT. Now relies on
the configurer default and only downgrades on explicit SASL_PLAINTEXT.
- Kerberos always throws: buildJaasConfig() called validateKerberos() before
checking if principal was set. Now returns null early for external JAAS
config.
- Missing callback handlers: OAuth and AWS MSK IAM never set required
callback
handler classes. Added getSaslCallbackProperties() helper with putIfAbsent
semantics. Also fixed OAuth scope key and moved token endpoint URL to
top-level
Kafka property.
Closes #24690
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../camel/component/kafka/KafkaConfiguration.java | 18 ++---
.../kafka/security/KafkaSecurityConfigurer.java | 40 ++++++++++-
.../camel/component/kafka/KafkaComponentTest.java | 62 ++++++++++++++++
.../security/KafkaSecurityConfigurerTest.java | 83 +++++++++++++++++++++-
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 16 +++++
5 files changed, 207 insertions(+), 12 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 57372b08af62..1e0ec5d2d957 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
@@ -449,15 +449,17 @@ public class KafkaConfiguration implements Cloneable,
HeaderFilterStrategyAware
// The saslAuthType just helps set the mechanism and protocol
}
- // Determine if we should use SSL (check if any SSL properties are
set)
- boolean hasSslConfig =
ObjectHelper.isNotEmpty(sslTruststoreLocation)
- || ObjectHelper.isNotEmpty(sslKeystoreLocation)
- || sslContextParameters != null;
-
- // For SASL types, default to SSL unless explicitly using
SASL_PLAINTEXT
+ // For SASL types, the configurer defaults to useSsl=true
(SASL_SSL).
+ // Only downgrade to SASL_PLAINTEXT if the user explicitly
requested it.
if (saslAuthType.isSasl()) {
- boolean useSsl = !securityProtocol.equals("SASL_PLAINTEXT") &&
!securityProtocol.equals("PLAINTEXT");
- configurer.withSsl(useSsl || hasSslConfig || saslAuthType !=
KafkaAuthType.NONE);
+ if (securityProtocol.equals("SASL_PLAINTEXT")) {
+ configurer.withSsl(false);
+ }
+ } else if (saslAuthType == KafkaAuthType.NONE) {
+ boolean hasSslConfig =
ObjectHelper.isNotEmpty(sslTruststoreLocation)
+ || ObjectHelper.isNotEmpty(sslKeystoreLocation)
+ || sslContextParameters != null;
+ configurer.withSsl(hasSslConfig ||
securityProtocol.equals("SSL"));
}
// Apply the configuration
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java
index 7538bd3d4fc3..b445b8c0639e 100644
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.kafka.security;
+import java.util.LinkedHashMap;
+import java.util.Map;
import java.util.Objects;
import java.util.Properties;
@@ -23,6 +25,8 @@ import org.apache.camel.component.kafka.KafkaConfiguration;
import org.apache.camel.util.ObjectHelper;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Builder for Kafka security configuration.
@@ -67,6 +71,8 @@ import org.apache.kafka.common.config.SaslConfigs;
*/
public class KafkaSecurityConfigurer {
+ private static final Logger LOG =
LoggerFactory.getLogger(KafkaSecurityConfigurer.class);
+
private final KafkaAuthType authType;
private String username;
private String password;
@@ -336,9 +342,8 @@ public class KafkaSecurityConfigurer {
validateOAuth();
sb.append("
clientId=\"").append(escapeJaasValue(oauthClientId)).append("\"");
sb.append("
clientSecret=\"").append(escapeJaasValue(oauthClientSecret)).append("\"");
- sb.append("
oauth.token.endpoint.uri=\"").append(oauthTokenEndpointUri).append("\"");
if (ObjectHelper.isNotEmpty(oauthScope)) {
- sb.append("
oauth.scope=\"").append(oauthScope).append("\"");
+ sb.append("
scope=\"").append(escapeJaasValue(oauthScope)).append("\"");
}
break;
@@ -347,6 +352,11 @@ public class KafkaSecurityConfigurer {
break;
case KERBEROS:
+ if (ObjectHelper.isEmpty(kerberosPrincipal)) {
+ LOG.debug(
+ "Kerberos principal not set; assuming external
JAAS configuration (e.g. java.security.auth.login.config)");
+ return null;
+ }
validateKerberos();
sb.append(" useKeyTab=").append(kerberosUseKeyTab);
sb.append(" storeKey=").append(kerberosStoreKey);
@@ -405,7 +415,32 @@ public class KafkaSecurityConfigurer {
if (jaasConfig != null) {
configuration.setSaslJaasConfig(jaasConfig);
}
+ applySaslProperties(configuration.getAdditionalProperties());
+ }
+ }
+
+ private void applySaslProperties(Map<String, Object> additionalProperties)
{
+ getSaslCallbackProperties().forEach(additionalProperties::putIfAbsent);
+ }
+
+ private Map<String, String> getSaslCallbackProperties() {
+ Map<String, String> props = new LinkedHashMap<>();
+ switch (authType) {
+ case OAUTH:
+ props.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
+
"org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler");
+ if (ObjectHelper.isNotEmpty(oauthTokenEndpointUri)) {
+ props.put(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL,
oauthTokenEndpointUri);
+ }
+ break;
+ case AWS_MSK_IAM:
+ props.put(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS,
+
"software.amazon.msk.auth.iam.IAMClientCallbackHandler");
+ break;
+ default:
+ break;
}
+ return props;
}
/**
@@ -426,6 +461,7 @@ public class KafkaSecurityConfigurer {
if (jaasConfig != null) {
props.setProperty(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
}
+ getSaslCallbackProperties().forEach(props::setProperty);
}
return props;
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java
index f2622e32c552..15154285dbac 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java
@@ -36,6 +36,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -420,6 +421,8 @@ public class KafkaComponentTest {
assertEquals("SASL_SSL",
props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
assertEquals("AWS_MSK_IAM",
props.getProperty(SaslConfigs.SASL_MECHANISM));
assertTrue(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("IAMLoginModule"));
+ assertEquals("software.amazon.msk.auth.iam.IAMClientCallbackHandler",
+
props.getProperty(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS));
}
@Test
@@ -446,6 +449,65 @@ public class KafkaComponentTest {
assertNull(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG));
}
+ @Test
+ public void testSaslAuthTypePlainWithExplicitSaslPlaintext() {
+ String uri = "kafka:mytopic?brokers=broker1:12345&saslAuthType=PLAIN"
+ +
"&saslUsername=user&saslPassword=pass&securityProtocol=SASL_PLAINTEXT";
+
+ KafkaEndpoint endpoint = context.getEndpoint(uri, KafkaEndpoint.class);
+
+ Properties props =
endpoint.getConfiguration().createProducerProperties();
+ assertEquals("SASL_PLAINTEXT",
props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertEquals("PLAIN", props.getProperty(SaslConfigs.SASL_MECHANISM));
+ }
+
+ @Test
+ public void testSaslAuthTypeKerberosDoesNotThrow() {
+ String uri =
"kafka:mytopic?brokers=broker1:12345&saslAuthType=KERBEROS";
+
+ KafkaEndpoint endpoint = context.getEndpoint(uri, KafkaEndpoint.class);
+ assertEquals(KafkaAuthType.KERBEROS,
endpoint.getConfiguration().getSaslAuthType());
+
+ Properties props =
endpoint.getConfiguration().createProducerProperties();
+ assertEquals("SASL_SSL",
props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertEquals("GSSAPI", props.getProperty(SaslConfigs.SASL_MECHANISM));
+ assertNull(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG));
+ }
+
+ @Test
+ public void testSaslAuthTypeOAuth() {
+ String uri = "kafka:mytopic?brokers=broker1:12345&saslAuthType=OAUTH"
+ + "&oauthClientId=myClient&oauthClientSecret=mySecret"
+ +
"&oauthTokenEndpointUri=https://auth.example.com/token&oauthScope=kafka";;
+
+ KafkaEndpoint endpoint = context.getEndpoint(uri, KafkaEndpoint.class);
+ assertEquals(KafkaAuthType.OAUTH,
endpoint.getConfiguration().getSaslAuthType());
+
+ Properties props =
endpoint.getConfiguration().createProducerProperties();
+ assertEquals("SASL_SSL",
props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ assertEquals("OAUTHBEARER",
props.getProperty(SaslConfigs.SASL_MECHANISM));
+ assertNotNull(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG));
+
assertTrue(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("OAuthBearerLoginModule"));
+
assertTrue(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("clientId=\"myClient\""));
+
assertTrue(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("scope=\"kafka\""));
+
assertFalse(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("oauth.token.endpoint.uri"));
+
assertEquals("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler",
+
props.getProperty(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ assertEquals("https://auth.example.com/token";,
+
props.getProperty(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL));
+ }
+
+ @Test
+ public void testSaslAuthTypeNoneWithExplicitSsl() {
+ String uri =
"kafka:mytopic?brokers=broker1:12345&saslAuthType=NONE&securityProtocol=SSL";
+
+ KafkaEndpoint endpoint = context.getEndpoint(uri, KafkaEndpoint.class);
+ assertEquals(KafkaAuthType.NONE,
endpoint.getConfiguration().getSaslAuthType());
+
+ Properties props =
endpoint.getConfiguration().createProducerProperties();
+ assertEquals("SSL",
props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+ }
+
// ========================================================================
// Backward compatibility tests - existing security options should still
work
// ========================================================================
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurerTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurerTest.java
index b70ad3449bc4..f2de4a435fe6 100644
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurerTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurerTest.java
@@ -231,8 +231,9 @@ public class KafkaSecurityConfigurerTest {
assertTrue(jaasConfig.contains("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule
required"));
assertTrue(jaasConfig.contains("clientId=\"my-client\""));
assertTrue(jaasConfig.contains("clientSecret=\"my-secret\""));
-
assertTrue(jaasConfig.contains("oauth.token.endpoint.uri=\"https://auth.example.com/token\"";));
- assertTrue(jaasConfig.contains("oauth.scope=\"kafka\""));
+ assertTrue(jaasConfig.contains("scope=\"kafka\""));
+ assertFalse(jaasConfig.contains("oauth.token.endpoint.uri"));
+ assertFalse(jaasConfig.contains("oauth.scope"));
}
@Test
@@ -466,6 +467,84 @@ public class KafkaSecurityConfigurerTest {
assertNull(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG));
}
+ // ========================================================================
+ // Kerberos without principal (external JAAS) tests
+ // ========================================================================
+
+ @Test
+ public void testBuildJaasConfigKerberosNoPrincipal() {
+ KafkaSecurityConfigurer configurer =
KafkaSecurityConfigurer.forAuthType(KafkaAuthType.KERBEROS);
+ assertNull(configurer.buildJaasConfig());
+ }
+
+ @Test
+ public void testKerberosNoPrincipalSetsProtocolAndMechanism() {
+ KafkaConfiguration config = new KafkaConfiguration();
+ KafkaSecurityConfigurer.forAuthType(KafkaAuthType.KERBEROS)
+ .configure(config);
+
+ assertEquals("SASL_SSL", config.getSecurityProtocol());
+ assertEquals("GSSAPI", config.getSaslMechanism());
+ assertNull(config.getSaslJaasConfig());
+ }
+
+ // ========================================================================
+ // Callback handler and additional SASL properties tests
+ // ========================================================================
+
+ @Test
+ public void testToPropertiesOAuthIncludesCallbackHandler() {
+ Properties props = KafkaSecurityConfigurer.oauth("client", "secret",
"https://auth.example.com/token";)
+ .toProperties();
+
+
assertEquals("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler",
+
props.getProperty(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ assertEquals("https://auth.example.com/token";,
+
props.getProperty(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL));
+ }
+
+ @Test
+ public void testToPropertiesAwsMskIamIncludesCallbackHandler() {
+ Properties props = KafkaSecurityConfigurer.awsMskIam().toProperties();
+
+ assertEquals("software.amazon.msk.auth.iam.IAMClientCallbackHandler",
+
props.getProperty(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS));
+ }
+
+ @Test
+ public void testConfigureOAuthSetsCallbackHandler() {
+ KafkaConfiguration config = new KafkaConfiguration();
+ KafkaSecurityConfigurer.oauth("client", "secret",
"https://auth.example.com/token";)
+ .configure(config);
+
+
assertEquals("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler",
+
config.getAdditionalProperties().get(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ assertEquals("https://auth.example.com/token";,
+
config.getAdditionalProperties().get(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL));
+ }
+
+ @Test
+ public void testConfigureAwsMskIamSetsCallbackHandler() {
+ KafkaConfiguration config = new KafkaConfiguration();
+ KafkaSecurityConfigurer.awsMskIam()
+ .configure(config);
+
+ assertEquals("software.amazon.msk.auth.iam.IAMClientCallbackHandler",
+
config.getAdditionalProperties().get(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS));
+ }
+
+ @Test
+ public void testConfigureOAuthDoesNotOverrideUserCallbackHandler() {
+ KafkaConfiguration config = new KafkaConfiguration();
+
config.getAdditionalProperties().put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
"com.custom.MyHandler");
+
+ KafkaSecurityConfigurer.oauth("client", "secret",
"https://auth.example.com/token";)
+ .configure(config);
+
+ assertEquals("com.custom.MyHandler",
+
config.getAdditionalProperties().get(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS));
+ }
+
// ========================================================================
// toString() test
// ========================================================================
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index e58517c676e4..935ef84e5273 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -240,6 +240,22 @@ but is applied to both consumer and producer Kafka
clients. The label has been c
If you use the Endpoint DSL, the `metadataMaxAgeMs` method has moved from the
producer (advanced) builder
to the common builder.
+=== camel-kafka - saslAuthType behavior changes
+
+When `saslAuthType` is set, the generated JAAS configuration and additional
SASL properties are now
+applied using `putIfAbsent` semantics, so explicitly configured
`saslJaasConfig`, callback handler
+classes, or token endpoint URLs take precedence over the auto-generated values.
+
+For OAuth (`saslAuthType=OAUTH`), the JAAS string now contains only
login-module parameters
+(`clientId`, `clientSecret`, `scope`). The `oauth.token.endpoint.uri` and
callback handler
+class are set as separate Kafka client properties instead of being embedded in
the JAAS string,
+matching the Kafka client's expected configuration format.
+
+For Kerberos (`saslAuthType=KERBEROS`), if the principal is not configured,
the configurer now
+assumes an external JAAS configuration (e.g. via
`java.security.auth.login.config`) and skips
+JAAS string generation instead of throwing an exception. This supports
deployment scenarios
+where Kerberos is configured externally.
+
=== camel-kafka - Producer no longer silently drops scalar Jackson nodes
With `useIterator=true`, the Kafka producer splits an `Iterable` body into one
record per element.
