无法使用Keycloak授权启动Kafka服务器

问：

服务器属性

security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
sasl.enabled.mechanisms=OAUTHBEARER
listeners=SASL_PLAINTEXT://localhost:9092
advertised.listeners=SASL_PLAINTEXT://localhost:9092
#Authorizer for ACL
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:kafka-client;
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

listener.name.sasl_plaintext.oauthbearer.sasl.login.callback.handler.class=com.duminda.security.oauthbearer.OauthAuthenticateLoginCallbackHandler
listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=com.duminda.security.oauthbearer.OauthAuthenticateValidatorCallbackHandler
sasl.login.callback.handler.class=com.duminda.security.oauthbearer.OauthAuthenticateLoginCallbackHandler
sasl.server.callback.handler.class=com.duminda.security.oauthbearer.OauthAuthenticateValidatorCallbackHandler

我使用了 https://dumisblog.wordpress.com/2020/06/04/kafka-security-sasl-oauthbearer-setup-with-keycloak/ 的 oauth 实现

当 Kafka 服务器启动时，它会启动，但显示节点已断开连接。

INFO ====active==== (com.duminda.security.oauthbearer.OauthHttpCalls) [2023-11-17 11:13:59,053] INFO Trying to introspected (com.duminda.security.oauthbearer.OauthAuthenticateValidatorCallbackHandler) [2023-11-17 11:13:59,054] INFO Validated! token.. (com.duminda.security.oauthbearer.OauthAuthenticateValidatorCallbackHandler) [2023-11-17 11:13:59,054] INFO [Controller id=1001, targetBrokerId=1001] Node 1001 disconnected. (org.apache.kafka.clients.NetworkClient) [2023-11-17 11:13:59,055] INFO [Controller id=1001, targetBrokerId=1001] Cancelled in-flight UPDATE_METADATA request with correlation id 671 due to node 1001 being disconnected (elapsed time since creation: 0ms, elapsed time since send: 0ms, request timeout: 30000ms) (org.apache.kafka.clients.NetworkClient) [2023-11-17 11:13:59,055] INFO [Controller id=1001, targetBrokerId=1001] Client requested connection close from node 1001 (org.apache.kafka.clients.NetworkClient)