我在Cloudera运行了3个经纪人kerberised Kafka 0.10安装,我正在尝试使用SASL / PLAIN进行身份验证
我将kafka_server_jaas.conf传递给每个代理的JVM .
KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username=admin password=password1 user_admin=password1 user_remote=password1; };
我的 server.properties
(或Cloudera重命名的 kafka.properties
)设置如下;
listeners=SASL_SSL://10.10.3.47:9093 # ip set for each broker advertised.listeners=SASL_SSL://10.10.3.47:9093 # ip set for each broker sasl.enabled.mechanisms=GSSAPI,PLAIN security.inter.broker.protocol=SASL_SSL sasl.mechanism.inter.broker.protocol=GSSAPI
当Kafka启动时,代理间通信一切正常,但是当我尝试使用控制台 生产环境 者进行连接时,我得到一个Timeout无法更新元数据
bin/kafka-consolproducer --broker-list 10.10.3.161:9093 --topic test1 --producer.config client.properties.plain
client.properties.plain设置为
security.protocol=SASL_SSL sasl.mechanism=PLAIN
最后,客户端jaas.conf
KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="remote" password="password1"; };
据我所知,我已经正确地遵循了所有指示,任何人都可以看到任何错误吗?
Update 我've turned the logging on the console consumer up a bit, I' m收到以下错误;
[2017-03-02 13:17:50,817] TRACE SSLHandshake NEED_UNWRAP channelId -1, handshakeResult Status = OK HandshakeStatus = FINISHED bytesConsumed = 101 bytesProduced = 0, appReadBuffer pos 0, netReadBuffer pos 0, netWriteBuffer pos 101 (org.apache.kafka.common.network.SslTransportLayer) [2017-03-02 13:17:50,817] TRACE SSLHandshake FINISHED channelId -1, appReadBuffer pos 0, netReadBuffer pos 0, netWriteBuffer pos 101 (org.apache.kafka.common.network.SslTransportLayer) [2017-03-02 13:17:50,817] DEBUG Set SASL client state to RECEIVE_HANDSHAKE_RESPONSE (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [2017-03-02 13:17:50,818] DEBUG Set SASL client state to INITIAL (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [2017-03-02 13:17:50,819] DEBUG Set SASL client state to INTERMEDIATE (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator) [2017-03-02 13:17:50,820] DEBUG Connection with <IPADDESS_REMOVED> disconnected (org.apache.kafka.common.network.Selector) java.io.EOFException at org.apache.kafka.common.network.SslTransportLayer.read(SslTransportLayer.java:488) at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:81) at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71) at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveResponseOrToken(SaslClientAuthenticator.java:239) at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:182) at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:64) at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:318) at org.apache.kafka.common.network.Selector.poll(Selector.java:283) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:183) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:974) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:938) at kafka.consumer.NewShinyConsumer.<init>(BaseConsumer.scala:61) at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:64) at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:51) at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala) [2017-03-02 13:17:50,821] DEBUG Node -1 disconnected. (org.apache.kafka.clients.NetworkClient)
1 回答
我在SASL_PLAINTEXT身份验证中遇到了类似的问题 . 我能够连接到代理(通过kafka-python),但是我从 生产环境 者发送的任何消息都会超时 .
我最后宣传了SASL_PLAINTEXT和PLAINTEXT监听器,但只是通过AWS安全组公开了SASL_PLAINTEXT监听器 .
我的server_jaas.conf基本相同 .
我的server.properties使用了以下设置:
我用kafka-python客户端调试它,我的命令看起来像这样(python)
通过此设置,我能够进行用户名/密码身份验证,并且还可以在不超时的情况下向代理生成和使用消息 .
希望这在某种程度上有所帮助:)