I am trying to configure the kafka sasl with mtls support. This my server property

############## General Broker Configuration ##############
broker.id=1
log.dirs=/root/kafka/kafka-sasl-ssl/kafka-logs


# Enable SASL + SSL for security
listeners=SASL_SSL://x.x.x.x:9093
advertised.listeners=SASL_SSL://x.x.x.x:9093

# Inter-broker protocol (even for single broker, should match client settings)
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

# JAAS Configuration for SASL (SCRAM Authentication)
listener.name.sasl_ssl.scram-sha-512.sasl.jaas.config=org.apache.kafkamon.security.scram.ScramLoginModule required username="user" password="user@321";

# Super users
super.users=User:user
sasl.enabled.mechanisms=SCRAM-SHA-512

############## SSL Configuration ##############
# Broker's Keystore (Server Certificate)
ssl.keystore.location=/root/kafka/kafka-sasl-ssl/certs/out/kafka.server.keystore.jks
ssl.keystore.password=password
ssl.key.password=password

# Broker's Truststore (CA for client verification)
ssl.truststore.location=/root/kafka/kafka-sasl-ssl/certs/out/kafka.server.truststore.jks
ssl.truststore.password=password

# Enable mTLS (Mutual Authentication) - Require client certificates
listener.name.sasl_ssl.ssl.client.auth=required

# Optional: Disable hostname verification if needed (for local testing)
ssl.endpoint.identification.algorithm=

############## Kafka Internal Topics Replication ##############
auto.create.topics.enable=false
default.replication.factor=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############## ZooKeeper Configuration ##############
zookeeper.connect=localhost:2182

I have given listener.name.sasl_ssl.ssl.client.auth=required for mtls. When I start the broker I am getting following error

javax.ssl|DEBUG|42|data-plane-kafka-network-thread-1-ListenerName(SASL_SSL)-SASL_SSL-0|2025-02-05 04:02:03.007 PST|SSLCipher.java:2024|KeyLimit write side: algorithm = AES/GCM/NOPADDING:KEYUPDATE
countdown value = 137438953472
javax.ssl|DEBUG|42|data-plane-kafka-network-thread-1-ListenerName(SASL_SSL)-SASL_SSL-0|2025-02-05 04:02:03.072 PST|ChangeCipherSpec.java:246|Consuming ChangeCipherSpec message
javax.ssl|DEBUG|42|data-plane-kafka-network-thread-1-ListenerName(SASL_SSL)-SASL_SSL-0|2025-02-05 04:02:03.073 PST|CertificateMessage.java:1172|Consuming client Certificate handshake message (
"Certificate": {
  "certificate_request_context": "",
  "certificate_list": [  
]
}
)
javax.ssl|ERROR|42|data-plane-kafka-network-thread-1-ListenerName(SASL_SSL)-SASL_SSL-0|2025-02-05 04:02:03.074 PST|TransportContext.java:375|Fatal (BAD_CERTIFICATE): Empty client certificate chain (
"throwable" : {
  javax.ssl.SSLHandshakeException: Empty client certificate chain
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:131)
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:117)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:370)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:326)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:317)
        at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.onConsumeCertificate(CertificateMessage.java:1188)
        at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.consume(CertificateMessage.java:1175)
        at java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:396)
        at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:480)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1277)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1264)
        at java.base/java.security.AccessController.doPrivileged(AccessController.java:712)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1209)
        at org.apache.kafkamonwork.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:443)
        at org.apache.kafkamonwork.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:532)
        at org.apache.kafkamonwork.SslTransportLayer.doHandshake(SslTransportLayer.java:381)
        at org.apache.kafkamonwork.SslTransportLayer.handshake(SslTransportLayer.java:301)
        at org.apache.kafkamonwork.KafkaChannel.prepare(KafkaChannel.java:178)
        at org.apache.kafkamonwork.Selector.pollSelectionKeys(Selector.java:547)
        at org.apache.kafkamonwork.Selector.poll(Selector.java:485)
        at kafkawork.Processor.poll(SocketServer.scala:1111)
        at kafkawork.Processor.run(SocketServer.scala:1015)
        at java.base/java.lang.Thread.run(Thread.java:840)}

)
javax.ssl|WARNING|42|data-plane-kafka-network-thread-1-ListenerName(SASL_SSL)-SASL_SSL-0|2025-02-05 04:02:03.075 PST|SSLEngineOutputRecord.java:182|outbound has closed, ignore outbound application data
[2025-02-05 04:02:03,076] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1] Failed authentication with /x.x.x.x (channelId=x.x.x.x:9093-x.x.x.x:64005-4) (SSL handshake failed) (org.apache.kafkamonwork.Selector)

Error states that empty client certificate, but I am just starting the broker/ server. But when using listener.name.sasl_ssl.ssl.client.auth=requested there are no errors.

Ref: #content/view/165224608

Complete working solution