Chapter 12. KafkaListenerAuthenticationCustom schema reference Used in: GenericKafkaListener Full list of KafkaListenerAuthenticationCustom schema properties To configure custom authentication, set the type property to custom . Custom authentication allows for any type of Kafka-supported authentication to be used. Example custom OAuth authentication configuration spec: kafka: config: principal.builder.class: SimplePrincipal.class listeners: - name: oauth-bespoke port: 9093 type: internal tls: true authentication: type: custom sasl: true listenerConfig: oauthbearer.sasl.client.callback.handler.class: client.class oauthbearer.sasl.server.callback.handler.class: server.class oauthbearer.sasl.login.callback.handler.class: login.class oauthbearer.connections.max.reauth.ms: 999999999 sasl.enabled.mechanisms: oauthbearer oauthbearer.sasl.jaas.config: | org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ; secrets: - name: example A protocol map is generated that uses the sasl and tls values to determine which protocol to map to the listener. SASL = True, TLS = True SASL_SSL SASL = False, TLS = True SSL SASL = True, TLS = False SASL_PLAINTEXT SASL = False, TLS = False PLAINTEXT 12.1. listenerConfig Listener configuration specified using listenerConfig is prefixed with listener.name. <listener_name>-<port> . For example, sasl.enabled.mechanisms becomes listener.name. <listener_name>-<port> .sasl.enabled.mechanisms . 12.2. secrets Secrets are mounted to /opt/kafka/custom-authn-secrets/custom-listener- <listener_name>-<port> / <secret_name> in the Kafka broker nodes' containers. For example, the mounted secret ( example ) in the example configuration would be located at /opt/kafka/custom-authn-secrets/custom-listener-oauth-bespoke-9093/example . 12.3. Principal builder You can set a custom principal builder in the Kafka cluster configuration. However, the principal builder is subject to the following requirements: The specified principal builder class must exist on the image. Before building your own, check if one already exists. You'll need to rebuild the Streams for Apache Kafka images with the required classes. No other listener is using oauth type authentication. This is because an OAuth listener appends its own principle builder to the Kafka configuration. The specified principal builder is compatible with Streams for Apache Kafka. Custom principal builders must support peer certificates for authentication, as Streams for Apache Kafka uses these to manage the Kafka cluster. Note Kafka's default principal builder class supports the building of principals based on the names of peer certificates. The custom principal builder should provide a principal of type user using the name of the SSL peer certificate. The following example shows a custom principal builder that satisfies the OAuth requirements of Streams for Apache Kafka. Example principal builder for custom OAuth configuration public final class CustomKafkaPrincipalBuilder implements KafkaPrincipalBuilder { public KafkaPrincipalBuilder() {} @Override public KafkaPrincipal build(AuthenticationContext context) { if (context instanceof SslAuthenticationContext) { SSLSession sslSession = ((SslAuthenticationContext) context).session(); try { return new KafkaPrincipal( KafkaPrincipal.USER_TYPE, sslSession.getPeerPrincipal().getName()); } catch (SSLPeerUnverifiedException e) { throw new IllegalArgumentException("Cannot use an unverified peer for authentication", e); } } // Create your own KafkaPrincipal here ... } } 12.4. KafkaListenerAuthenticationCustom schema properties The type property is a discriminator that distinguishes use of the KafkaListenerAuthenticationCustom type from KafkaListenerAuthenticationTls , KafkaListenerAuthenticationScramSha512 , KafkaListenerAuthenticationOAuth . It must have the value custom for the type KafkaListenerAuthenticationCustom . Property Property type Description listenerConfig map Configuration to be used for a specific listener. All values are prefixed with listener.name. <listener_name> . sasl boolean Enable or disable SASL on this listener. secrets GenericSecretSource array Secrets to be mounted to /opt/kafka/custom-authn-secrets/custom-listener- <listener_name>-<port> / <secret_name> . type string Must be custom .

Commands:
spec: kafka: config: principal.builder.class: SimplePrincipal.class listeners: - name: oauth-bespoke port: 9093 type: internal tls: true authentication: type: custom sasl: true listenerConfig: oauthbearer.sasl.client.callback.handler.class: client.class oauthbearer.sasl.server.callback.handler.class: server.class oauthbearer.sasl.login.callback.handler.class: login.class oauthbearer.connections.max.reauth.ms: 999999999 sasl.enabled.mechanisms: oauthbearer oauthbearer.sasl.jaas.config: | org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ; secrets: - name: example
public final class CustomKafkaPrincipalBuilder implements KafkaPrincipalBuilder { public KafkaPrincipalBuilder() {} @Override public KafkaPrincipal build(AuthenticationContext context) { if (context instanceof SslAuthenticationContext) { SSLSession sslSession = ((SslAuthenticationContext) context).session(); try { return new KafkaPrincipal( KafkaPrincipal.USER_TYPE, sslSession.getPeerPrincipal().getName()); } catch (SSLPeerUnverifiedException e) { throw new IllegalArgumentException("Cannot use an unverified peer for authentication", e); } } // Create your own KafkaPrincipal here } }