Not working over SASL_SSL

[vadirajks]

Hi, Thanks for sharing your code and it works pretty well with SASL_PLAINTEXT.
Unfortunately, i am getting few errors by making use of SSL[SASL_SSL]. Please let me know if anything needs to corrected.

Enabling Java debug options: -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005
Listening for transport dt_socket at address: 5005
[2022-01-30 17:07:16,064] WARN [AdminClient clientId=adminclient-1] Metadata update failed due to authentication error (org.apache.kafka.clients.admin.internals.AdminMetadataManager)
org.apache.kafka.common.errors.SaslAuthenticationException: {"status":"invalid_token"}
[2022-01-30 17:07:16,067] ERROR [AdminClient clientId=adminclient-1] Connection to node -1 (kafka01.example.com/34.8.22.60:9095) failed authentication due to: {"status":"invalid_token"} (org.apache.kafka.clients.NetworkClient)
Error while executing topic command : org.apache.kafka.common.errors.SaslAuthenticationException: {"status":"invalid_token"}
[2022-01-30 17:07:16,073] ERROR java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SaslAuthenticationException: {"status":"invalid_token"}
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
at kafka.admin.TopicCommand$AdminClientTopicService.createTopic(TopicCommand.scala:163)
at kafka.admin.TopicCommand$TopicService.createTopic(TopicCommand.scala:134)
at kafka.admin.TopicCommand$TopicService.createTopic$(TopicCommand.scala:129)
at kafka.admin.TopicCommand$AdminClientTopicService.createTopic(TopicCommand.scala:157)
at kafka.admin.TopicCommand$.main(TopicCommand.scala:60)
at kafka.admin.TopicCommand.main(TopicCommand.scala)
Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: {"status":"invalid_token"}
(kafka.admin.TopicCommand$)

[2022-01-30 11:37:14,278] DEBUG Processor 10 listening to new connection from /116.75.126.142:60134 (kafka.network.Processor)
[2022-01-30 11:37:14,284] DEBUG connections.max.reauth.ms for mechanism=OAUTHBEARER: 0 (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2022-01-30 11:37:14,588] DEBUG [SslTransportLayer channelId=10.125.64.52:9095-116.75.126.142:60134-1 key=sun.nio.ch.SelectionKeyImpl@e3c9ee0] SSL peer is not authenticated, returning ANONYMOUS instead (org.apache.kafka.common.network.SslTransportLayer)
[2022-01-30 11:37:14,588] DEBUG [SslTransportLayer channelId=10.125.64.52:9095-116.75.126.142:60134-1 key=sun.nio.ch.SelectionKeyImpl@e3c9ee0] SSL handshake completed successfully with peerHost '116.75.126.142' peerPort 60134 peerPrincipal 'User:ANONYMOUS' cipherSuite 'TLS_DHE_DSS_WITH_AES_128_CBC_SHA256' (org.apache.kafka.common.network.SslTransportLayer)
[2022-01-30 11:37:14,846] DEBUG Set SASL server state to HANDSHAKE_OR_VERSIONS_REQUEST during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2022-01-30 11:37:14,846] DEBUG Handling Kafka request API_VERSIONS during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2022-01-30 11:37:14,847] DEBUG Set SASL server state to HANDSHAKE_REQUEST during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2022-01-30 11:37:15,101] DEBUG Handling Kafka request SASL_HANDSHAKE during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2022-01-30 11:37:15,101] DEBUG Using SASL mechanism 'OAUTHBEARER' provided by client (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2022-01-30 11:37:15,103] DEBUG Set SASL server state to AUTHENTICATE during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2022-01-30 11:37:15,117] DEBUG Got ping response for sessionid: 0x2003c00c4ca0021 after 0ms (org.apache.zookeeper.ClientCnxn)
[2022-01-30 11:37:15,360] DEBUG {"status":"invalid_token"} (org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslServer)
[2022-01-30 11:37:15,610] DEBUG Received %x01 response from client after it received our error (org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslServer)
[2022-01-30 11:37:15,610] DEBUG Set SASL server state to FAILED during authentication (org.apache.kafka.common.security.authenticator.SaslServerAuthenticator)
[2022-01-30 11:37:15,611] INFO [SocketServer brokerId=1] Failed authentication with 116.75.126.142/116.75.126.142 ({"status":"invalid_token"}) (org.apache.kafka.common.network.Selector)