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

Problem

Knative Eventing's Kafka broker cannot authenticate to Azure Event Hubs using Microsoft Entra ID (Managed Identity / Workload Identity). Only SAS connection strings work today, which is a blocker in environments where static shared secrets are prohibited.

Root cause: When sasl.mechanism=OAUTHBEARER, KafkaClientsAuth.java returns early without setting sasl.jaas.config or sasl.login.callback.handler.class. The Kafka client is left with the default OAuthBearerLoginModule which has no way to obtain a token.

This is deliberate for AWS MSK IAM, where IAMLoginModule is bundled on the classpath. But there is no equivalent for Azure, and no way to point the client at a custom handler through the auth secret.

Why this is not an Azure-specific problem

Azure Event Hubs uses standard OAUTHBEARER — the only missing piece is letting operators declare which AuthenticateCallbackHandler to use. The same gap blocks:

• Strimzi + Keycloak OIDC
• Confluent Cloud OIDC
• Any broker implementing standard OAUTHBEARER

Evidence: end-to-end PoC with zero static credentials

We validated the approach by deploying a derived data-plane image with the Conduktor Azure OAUTHBEARER handler on the classpath, and configuring the auth via ConfigMap properties (workaround for the missing secret propagation):

• ✅ End-to-end event flow through Broker → Event Hubs → Trigger
• ✅ Workload Identity federation (zero secrets in the cluster)
• ✅ Token refresh soak test: >2.5 hours continuous operation, zero auth failures, zero event loss across multiple token expiry cycles

Full PoC documentation: https://dmetzler.github.io/knative-on-azure/phase1-oauthbearer-poc/

Proposed fix

A vendor-neutral passthrough — no new dependencies, no Azure code:

1. Java data plane (KafkaClientsAuth.java): When sasl.mechanism=OAUTHBEARER, propagate sasl.jaas.config and sasl.login.callback.handler.class from the auth secret to the Kafka client properties when present. When absent → no-op (preserves MSK IAM behaviour).

2. Go control plane (secret.go): Tolerate OAUTHBEARER secrets without tokenProvider. The new keys are consumed by the Java data plane only; the Go side ignores them without error.

3. Credentials interface: Add SASLJaasConfig() and SASLLoginCallbackHandlerClass() default methods.

What this does NOT do

• ❌ No Azure SDK in the image
• ❌ No vendor-specific code of any kind
• ❌ No token minting/caching/refresh logic (that's the callback handler's job)
• ❌ No change to the AWS MSK IAM path (byte-identical when new keys are absent)

Secret format (new optional keys)

apiVersion: v1
kind: Secret
metadata:
  name: kafka-auth-secret
stringData:
  protocol: "SASL_SSL"
  sasl.mechanism: "OAUTHBEARER"
  sasl.jaas.config: "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required scope=\"https://<ns>.servicebus.windows.net/.default\";"
  sasl.login.callback.handler.class: "io.conduktor.kafka.security.oauthbearer.azure.AzureManagedIdentityCallbackHandler"

Known limitation

The Go control plane cannot use a Java callback handler. When tokenProvider is absent, it configures SASL/OAUTHBEARER as enabled but without a Go-native token provider. This is acceptable: the control plane manages configuration only; the data plane handles all Kafka traffic.

Implementation

Branch with passing tests (Java + Go): https://github.com/dmetzler/eventing-kafka-broker/tree/feature/oauthbearer-passthrough

Happy to open a PR once there's maintainer signal on the approach.