Chapter 12. KafkaListenerAuthenticationCustom schema reference

spec:
  kafka:
    listeners:
      - name: tls
        port: 9093
        tls: true
        type: internal
        authentication:
          type: custom
          sasl: false
          listenerConfig:
            ssl.client.auth: required
            ssl.principal.mapping.rules: RULE:^CN=(.*?),(.*)$/$1@my-cluster.com/
            ssl.truststore.location: /mnt/my-truststore/ca.crt
            ssl.truststore.type: PEM
    template:
      pod:
        volumes:
          - name: my-truststore
            secret:
              secretName: custom-truststore
      kafkaContainer:
        volumeMounts:
          - name: my-truststore
            mountPath: /mnt/my-truststore

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
        ...
    }
}