I have a use case where I need to configure separate authentication for inter-broker communication and another for producer and consumers.

For producer/consumer we have decided to go with SASL/OAUTHBEARER.

For this to happen I tried below as shown

## Configuration Overrides. Specify any Kafka settings you would like set on the StatefulSet
## here in map format, as defined in the official docs.
## ref: https://kafka.apache.org/documentation/#brokerconfigs
##
configurationOverrides:
  "offsets.topic.replication.factor": 1
  "confluent.support.metrics.enable": false  # Disables confluent metric submission
  "auto.create.topics.enable": false
  "delete.topic.enable": true
  "unclean.leader.election.enable": false
  "zookeeper.connect": zookeeper.zookeeper.svc.cluster.local:2181
  "zookeeper.sasl.enabled": false
  "super.users": User:admin
  "inter.broker.listener.name": EXTERNAL
  "sasl.mechanism.inter.broker.protocol": OAUTHBEARER
  "sasl.enabled.mechanisms": OAUTHBEARER
  "listener.name.external.oauthbearer.sasl.login.callback.handler.class": oracle.insight.common.kafka.security.OAuthBearerSignedLoginCallbackHandler
  "listener.name.external.oauthbearer.sasl.server.callback.handler.class": oracle.insight.common.kafka.security.OAuthBearerSignedValidatorCallbackHandler
  "listener.security.protocol.map": PLAINTEXT:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT
  "listener.name.external.oauthbearer.sasl.jaas.config": org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required signedLoginStringClaim_sub=<tenant-subject> signedLoginKeyServiceClass=oracle.insight.common.security.SMSKeyService signedValidatorKeyServiceClass=oracle.insight.common.security.SMSKeyService;
  "advertised.listeners": EXTERNAL://kafka-$((${KAFKA_BROKER_ID})).<domain-name>:$((${KAFKA_OUTSIDE_PORT} + ${KAFKA_BROKER_ID}))

With this approach, as assuming both inter-broker and producer/consumer are using the same auth i.e SASL/OAUTHBEARER and this works.

Below is the help am looking for;

1. For producer/consumer, I need only SASL/OAUTHBEARER validation in the broker side. So am thinking we might need to specify only below. I do not want to hardcode signedLoginStringClaim_sub. Producer with send the sub and the validator code is responsible for validating it.

"listener.name.external.oauthbearer.sasl.server.callback.handler.class": oracle.insight.common.kafka.security.OAuthBearerSignedValidatorCallbackHandler
"listener.security.protocol.map": PLAINTEXT:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT
"advertised.listeners": EXTERNAL://kafka-$((${KAFKA_BROKER_ID})).<domain-name>:$((${KAFKA_OUTSIDE_PORT} + ${KAFKA_BROKER_ID}))

This always fails with jaas.config not found.

2. How do I specify different listener with another auth mechanism for inter-broker communication?

Please help is solving 1. and 2. questions.

Thanks.