feat: propagate sasl.jaas.config and sasl.login.callback.handler.class from auth secret for OAUTHBEARER

control-plane/pkg/security/oauth/common.go

@@ -58,3 +58,9 @@ func getAWSRegion(data map[string][]byte) string {
 
 	return defaultAWSRegion
 }
+
+// HasTokenProvider returns true if the secret data contains a tokenProvider key.
+func HasTokenProvider(data map[string][]byte) bool {
+	_, ok := data[saslTokenProviderKey]
+	return ok
+}

control-plane/pkg/security/oauth/common_test.go

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2025 The Knative Authors
+ *
+ * Licensed 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.
+ */
+
+package oauth
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHasTokenProvider(t *testing.T) {
+	tests := []struct {
+		name string
+		data map[string][]byte
+		want bool
+	}{
+		{
+			name: "tokenProvider present",
+			data: map[string][]byte{saslTokenProviderKey: []byte(mskAccessTokenProvider)},
+			want: true,
+		},
+		{
+			name: "tokenProvider present but empty still counts as set",
+			data: map[string][]byte{saslTokenProviderKey: []byte("")},
+			want: true,
+		},
+		{
+			name: "tokenProvider absent",
+			data: map[string][]byte{saslAWSRegion: []byte(defaultAWSRegion)},
+			want: false,
+		},
+		{
+			name: "empty data",
+			data: map[string][]byte{},
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, HasTokenProvider(tt.data))
+		})
+	}
+}

control-plane/pkg/security/oauth/token_provider.go

@@ -65,3 +65,22 @@ func NewTokenProvider(data map[string][]byte) (*TokenProvider, error) {
 		tokenIssuer: tokenIssuer,
 	}, nil
 }
+
+// errorTokenProvider is a sarama.AccessTokenProvider that always fails with a
+// descriptive error. It is used when an OAUTHBEARER secret has no tokenProvider
+// (the sasl.jaas.config / sasl.login.callback.handler.class keys are consumed by
+// the Java data plane only). Setting this instead of leaving TokenProvider nil
+// prevents a nil-pointer panic in sarama's OAUTHBEARER auth path when the Go
+// control plane attempts an admin connection, turning it into a clear reconciler
+// error instead.
+type errorTokenProvider struct{ msg string }
+
+func (e *errorTokenProvider) Token() (*sarama.AccessToken, error) {
+	return nil, fmt.Errorf("%s", e.msg)
+}
+
+// UnsupportedTokenProvider returns an AccessTokenProvider that always fails with
+// the given message. See errorTokenProvider for rationale.
+func UnsupportedTokenProvider(msg string) sarama.AccessTokenProvider {
+	return &errorTokenProvider{msg: msg}
+}

control-plane/pkg/security/oauth/token_provider_test.go

@@ -139,3 +139,15 @@ func TestNewTokenProvider(t *testing.T) {
 		})
 	}
 }
+
+func TestUnsupportedTokenProvider(t *testing.T) {
+	const msg = "OAUTHBEARER without a Go token provider is not supported by the control plane"
+
+	provider := UnsupportedTokenProvider(msg)
+	assert.NotNil(t, provider)
+
+	token, err := provider.Token()
+	assert.Nil(t, token)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), msg)
+}

control-plane/pkg/security/secret.go

@@ -46,6 +46,11 @@ const (
 	SaslTypeLegacy   = "saslType" // legacy secrets
 	SaslUsernameKey  = "username" // legacy secrets
 
+	// OAUTHBEARER passthrough keys — used by the Java data plane only.
+	// The Go control plane tolerates and ignores them.
+	SaslJaasConfigKey                = "sasl.jaas.config"
+	SaslLoginCallbackHandlerClassKey = "sasl.login.callback.handler.class"
+
 	ProtocolPlaintext     = "PLAINTEXT"
 	ProtocolSASLPlaintext = "SASL_PLAINTEXT"
 	ProtocolSSL           = "SSL"
@@ -116,11 +121,24 @@ func saslConfig(protocol string, data map[string][]byte) kafka.ConfigOption {
 		if saslMechanism == SaslOAuth {
 			config.Net.SASL.Enable = true
 			config.Net.SASL.Mechanism = sarama.SASLTypeOAuth
-			tokenProvider, err := oauth.NewTokenProvider(data)
-			if err != nil {
-				return fmt.Errorf("[protocol %s] failed to create OAUTHBEARER token provider: %w", protocol, err)
+
+			// If the secret contains a tokenProvider key (MSK), use the Go-native token provider.
+			// If not, the secret likely contains sasl.jaas.config / sasl.login.callback.handler.class
+			// for the Java data plane (Azure, Keycloak, etc.). The control plane cannot use a Java
+			// callback handler, so we configure SASL/OAUTHBEARER as enabled but skip the token
+			// provider — the control plane will not produce/consume from Kafka in this mode.
+			if oauth.HasTokenProvider(data) {
+				tokenProvider, err := oauth.NewTokenProvider(data)
+				if err != nil {
+					return fmt.Errorf("[protocol %s] failed to create OAUTHBEARER token provider: %w", protocol, err)
+				}
+				config.Net.SASL.TokenProvider = tokenProvider
+			} else {
+				config.Net.SASL.TokenProvider = oauth.UnsupportedTokenProvider(
+					"OAUTHBEARER without tokenProvider is not supported by the Go control plane " +
+						"(sasl.jaas.config/sasl.login.callback.handler.class are consumed by the Java data plane only); " +
+						"topic and consumer group management will fail")
 			}
-			config.Net.SASL.TokenProvider = tokenProvider
 			return nil
 		}
 

control-plane/pkg/security/secret_test.go

@@ -209,6 +209,9 @@ func TestSASLOAuthWithMSKRoleProvider(t *testing.T) {
 }
 
 func TestSASLOAuthMissingTokenProvider(t *testing.T) {
+	// When tokenProvider is absent, OAUTHBEARER is still valid — the secret may contain
+	// sasl.jaas.config / sasl.login.callback.handler.class for the Java data plane.
+	// The Go control plane tolerates this and configures SASL as enabled without a token provider.
 	secret := map[string][]byte{
 		"protocol":       []byte("SASL_PLAINTEXT"),
 		"sasl.mechanism": []byte("OAUTHBEARER"),
@@ -217,8 +220,13 @@ func TestSASLOAuthMissingTokenProvider(t *testing.T) {
 
 	err := kafka.Options(config, secretData(secret))
 
-	assert.NotNil(t, err)
-	assert.Contains(t, err.Error(), "OAUTHBEARER token provider required")
+	assert.Nil(t, err)
+	assert.True(t, config.Net.SASL.Enable)
+	assert.Equal(t, sarama.SASLMechanism(sarama.SASLTypeOAuth), config.Net.SASL.Mechanism)
+	assert.NotNil(t, config.Net.SASL.TokenProvider)
+	_, tokenErr := config.Net.SASL.TokenProvider.Token()
+	assert.NotNil(t, tokenErr)
+	assert.Contains(t, tokenErr.Error(), "not supported by the Go control plane")
 }
 
 func TestSASLOAuthInvalidTokenProvider(t *testing.T) {
@@ -235,6 +243,30 @@ func TestSASLOAuthInvalidTokenProvider(t *testing.T) {
 	assert.Contains(t, err.Error(), "unsupported OAUTHBEARER token provider")
 }
 
+func TestSASLOAuthWithJavaPassthroughKeys(t *testing.T) {
+	// When the secret contains sasl.jaas.config and sasl.login.callback.handler.class
+	// but no tokenProvider, the Go control plane should accept it without error.
+	// These keys are consumed by the Java data plane only.
+	secret := map[string][]byte{
+		"protocol":                          []byte("SASL_SSL"),
+		"sasl.mechanism":                    []byte("OAUTHBEARER"),
+		"sasl.jaas.config":                  []byte("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;"),
+		"sasl.login.callback.handler.class": []byte("io.conduktor.kafka.security.oauthbearer.azure.AzureManagedIdentityCallbackHandler"),
+	}
+	config := sarama.NewConfig()
+
+	err := kafka.Options(config, secretData(secret))
+
+	assert.Nil(t, err)
+	assert.True(t, config.Net.SASL.Enable)
+	assert.Equal(t, sarama.SASLMechanism(sarama.SASLTypeOAuth), config.Net.SASL.Mechanism)
+	assert.NotNil(t, config.Net.SASL.TokenProvider)
+	_, tokenErr := config.Net.SASL.TokenProvider.Token()
+	assert.NotNil(t, tokenErr)
+	assert.Contains(t, tokenErr.Error(), "not supported by the Go control plane")
+	assert.True(t, config.Net.TLS.Enable)
+}
+
 func TestSSL(t *testing.T) {
 	ca, userKey, userCert := loadCerts(t)
 

data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/Credentials.java

@@ -76,4 +76,30 @@ public interface Credentials {
      * @see <a href="https://kafka.apache.org/documentation/#security_sasl_scram">SASL Scram</a>
      */
     String SASLPassword();
+
+    /**
+     * Client key: sasl.jaas.config
+     *
+     * <p>When set, this value is used verbatim as the JAAS configuration for the Kafka client.
+     * Primarily useful for OAUTHBEARER, where the operator supplies a custom login module
+     * and its configuration (e.g. scope, token endpoint).
+     *
+     * @return the full JAAS config string, or null if not specified.
+     */
+    default String SASLJaasConfig() {
+        return null;
+    }
+
+    /**
+     * Client key: sasl.login.callback.handler.class
+     *
+     * <p>Fully qualified class name of an {@code AuthenticateCallbackHandler} implementation
+     * that the Kafka client will use to obtain tokens. The class must be on the data-plane
+     * classpath — this project does not ship any vendor-specific handler.
+     *
+     * @return the handler class name, or null if not specified.
+     */
+    default String SASLLoginCallbackHandlerClass() {
+        return null;
+    }
 }

data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java

@@ -24,9 +24,13 @@
 import org.apache.kafka.common.security.plain.PlainLoginModule;
 import org.apache.kafka.common.security.scram.ScramLoginModule;
 import org.apache.kafka.common.security.ssl.DefaultSslEngineFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class KafkaClientsAuth {
 
+    private static final Logger logger = LoggerFactory.getLogger(KafkaClientsAuth.class);
+
     public static Properties attachCredentials(final Properties properties, final Credentials credentials) {
         clientsProperties(properties::setProperty, credentials);
         return properties;
@@ -72,14 +76,33 @@ private static void sasl(final BiConsumer<String, String> propertiesSetter, fina
                             credentials.SASLUsername(),
                             credentials.SASLPassword()));
         } else if ("OAUTHBEARER".equals(mechanism)) {
-            // OAUTHBEARER does not require username and password, so we do not set them.
+            // Propagate JAAS config and callback handler class when provided in the secret.
+            // This enables vendor-neutral OAUTHBEARER auth (Azure Event Hubs, Keycloak, etc.)
+            // without adding any vendor dependency to this project.
+            // When neither key is present, this is a no-op — preserving the current behaviour
+            // for AWS MSK IAM, which supplies its own login module via the classpath.
+            final var jaasConfig = credentials.SASLJaasConfig();
+            if (jaasConfig != null) {
+                propertiesSetter.accept(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
+            }
+            final var callbackHandler = credentials.SASLLoginCallbackHandlerClass();
+            if (callbackHandler != null) {
+                propertiesSetter.accept(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, callbackHandler);
+            }
         } else {
             propertiesSetter.accept(
                     SaslConfigs.SASL_JAAS_CONFIG,
                     String.format(
                             ScramLoginModule.class.getName() + " required username=\"%s\" password=\"%s\";",
                             credentials.SASLUsername(),
                             credentials.SASLPassword()));
+            // Warn if callback handler class is set on a non-OAUTHBEARER mechanism — likely misconfiguration.
+            if (credentials.SASLLoginCallbackHandlerClass() != null) {
+                logger.warn(
+                        "sasl.login.callback.handler.class is set but sasl.mechanism is '{}', not OAUTHBEARER. "
+                                + "The callback handler class will be ignored.",
+                        mechanism);
+            }
         }
     }
 

data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentials.java

@@ -43,6 +43,8 @@ class KubernetesCredentials implements Credentials {
 
     static final String SECURITY_PROTOCOL = "protocol";
     static final String SASL_MECHANISM = "sasl.mechanism";
+    static final String SASL_JAAS_CONFIG_KEY = "sasl.jaas.config";
+    static final String SASL_LOGIN_CALLBACK_HANDLER_CLASS_KEY = "sasl.login.callback.handler.class";
 
     private final Secret secret;
 
@@ -54,6 +56,10 @@ class KubernetesCredentials implements Credentials {
     private String SASLMechanism;
     private String SASLUsername;
     private String SASLPassword;
+    private String SASLJaasConfig;
+    private String SASLLoginCallbackHandlerClass;
+    private boolean SASLJaasConfigResolved;
+    private boolean SASLLoginCallbackHandlerClassResolved;
 
     KubernetesCredentials(final Secret secret) {
         this.secret = secret;
@@ -200,4 +206,35 @@ public String SASLPassword() {
         }
         return this.SASLPassword;
     }
+
+    @Override
+    public String SASLJaasConfig() {
+        if (secret == null || secret.getData() == null) {
+            return null;
+        }
+        if (!SASLJaasConfigResolved) {
+            final var value = secret.getData().get(SASL_JAAS_CONFIG_KEY);
+            if (value != null) {
+                this.SASLJaasConfig = new String(Base64.getDecoder().decode(value));
+            }
+            this.SASLJaasConfigResolved = true;
+        }
+        return this.SASLJaasConfig;
+    }
+
+    @Override
+    public String SASLLoginCallbackHandlerClass() {
+        if (secret == null || secret.getData() == null) {
+            return null;
+        }
+        if (!SASLLoginCallbackHandlerClassResolved) {
+            final var value = secret.getData().get(SASL_LOGIN_CALLBACK_HANDLER_CLASS_KEY);
+            if (value != null) {
+                this.SASLLoginCallbackHandlerClass =
+                        new String(Base64.getDecoder().decode(value));
+            }
+            this.SASLLoginCallbackHandlerClassResolved = true;
+        }
+        return this.SASLLoginCallbackHandlerClass;
+    }
 }