我想做的是：

Setup SASL/PLAIN for inter-broker communication

Setup SASL/PLAIN for Broker-Zookeeper communication

And Setup SASL/OAUTHBEARER for client-broker communication.

我的服务器\u jaas.conf是：

KafkaServer { 
    org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required
    LoginStringClaim_sub="admin";

    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret";
};

Client {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="zookeeper"
    password="zookeeper_secret";
};

我还设置了以下配置选项：

sasl.enabled.mechanisms=PLAIN,OAUTHBEARER
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.server.callback.handler.class=br.com.jairsjunior.security.oauthbearer.OauthAuthenticateValidatorCallbackHandler

我用了这里提供的jar-https://github.com/jairsjunior/kafka-playground/tree/master/kafka-broker/kafka-with-oauth/libs
但是当我启动Kafka服务器的时候，我得到了一个错误：

Unrecognized SASL Login callback
(org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule)
javax.security.auth.callback.UnsupportedCallbackException: Unrecognized SASL Login callback

at org.apache.kafka.common.security.authenticator.AbstractLogin$DefaultLoginCallbackHandler.handle(AbstractLogin.java:105)
at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.login(OAuthBearerLoginModule.java:261)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at javax.security.auth.login.LoginContext.invoke(LoginContext.java:755)
at javax.security.auth.login.LoginContext.access$000(LoginContext.java:195)
at javax.security.auth.login.LoginContext$4.run(LoginContext.java:682)
at javax.security.auth.login.LoginContext$4.run(LoginContext.java:680)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.login.LoginContext.invokePriv(LoginContext.java:680)
at javax.security.auth.login.LoginContext.login(LoginContext.java:587)
at org.apache.kafka.common.security.authenticator.AbstractLogin.login(AbstractLogin.java:60)
at org.apache.kafka.common.security.authenticator.LoginManager.<init>(LoginManager.java:64)
at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:114)
at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:142)
at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:140)
at org.apache.kafka.common.network.ChannelBuilders.serverChannelBuilder(ChannelBuilders.java:82)
at kafka.network.Processor.<init>(SocketServer.scala:548)
at kafka.network.SocketServer.newProcessor(SocketServer.scala:247)
at kafka.network.SocketServer.$anonfun$addProcessors$1(SocketServer.scala:163)
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:156)
at kafka.network.SocketServer.addProcessors(SocketServer.scala:162)
at kafka.network.SocketServer.$anonfun$createAcceptorAndProcessors$1(SocketServer.scala:150)
at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:52)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at kafka.network.SocketServer.createAcceptorAndProcessors(SocketServer.scala:145)
at kafka.network.SocketServer.startup(SocketServer.scala:94)
at kafka.server.KafkaServer.startup(KafkaServer.scala:250)
at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:38)
at kafka.Kafka$.main(Kafka.scala:75)
at kafka.Kafka.main(Kafka.scala)

[2019-05-24 12:30:30,414] ERROR [KafkaServer id=240] Fatal error during   
KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
org.apache.kafka.common.KafkaException:    
javax.security.auth.login.LoginException: An internal error occurred
at
org.apache.kafka.common.network.SaslChannelBuilder.
configure(SaslChannelBuilder.java:153)
at org.apache.kafka.common.network.ChannelBuilders.create
(ChannelBuilders.java:140)
at org.apache.kafka.common.network.ChannelBuilders.serverChannelBuilder
(ChannelBuilders.java:82)
at kafka.network.Processor.<init>(SocketServer.scala:548)
at kafka.network.SocketServer.newProcessor(SocketServer.scala:247)
at kafka.network.SocketServer.$anonfun$addProcessors$1
(SocketServer.scala:163)
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:156)
at kafka.network.SocketServer.addProcessors(SocketServer.scala:162)
at kafka.network.SocketServer.$anonfun$createAcceptorAndProcessors$1
(SocketServer.scala:150)
at scala.collection.mutable.ResizableArray.foreach
(ResizableArray.scala:59)
at scala.collection.mutable.ResizableArray.foreach$
(ResizableArray.scala:52)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at kafka.network.SocketServer.createAcceptorAndProcessors
(SocketServer.scala:145)
at kafka.network.SocketServer.startup(SocketServer.scala:94)
at kafka.server.KafkaServer.startup(KafkaServer.scala:250)
at kafka.server.KafkaServerStartable.startup
(KafkaServerStartable.scala:38)
at kafka.Kafka$.main(Kafka.scala:75)
at kafka.Kafka.main(Kafka.scala)

Caused by: javax.security.auth.login.LoginException: An internal error occurred
at org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.
login(OAuthBearerLoginModule.java:264)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke
(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke
(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at javax.security.auth.login.LoginContext.invoke(LoginContext.java:755)
at javax.security.auth.login.LoginContext.access$000
(LoginContext.java:195)
at javax.security.auth.login.LoginContext$4.run(LoginContext.java:682)
at javax.security.auth.login.LoginContext$4.run(LoginContext.java:680)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.login.LoginContext.invokePriv
(LoginContext.java:680)
at javax.security.auth.login.LoginContext.login(LoginContext.java:587)
at org.apache.kafka.common.security.authenticator.AbstractLogin.
login(AbstractLogin.java:60)
at org.apache.kafka.common.security.authenticator.LoginManager.
<init>(LoginManager.java:64)
at org.apache.kafka.common.security.authenticator.LoginManager.
acquireLoginManager(LoginManager.java:114)
at org.apache.kafka.common.network.SaslChannelBuilder.configure
(SaslChannelBuilder.java:142)
... 17 more

我错过了什么？任何帮助都将不胜感激。
谢谢！