kafka after ssl produce and consume message failed

[#1]

After I configure Kafka security with SSL, I execute the command to produce and consume message, but it prints messages as follows:

[2017-05-16 06:45:20,660] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:20,937] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:21,087] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:21,403] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:21,629] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:21,776] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:21,932] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:22,155] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:22,305] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:22,451] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:22,602] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:22,751] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:22,958] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:23,104] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:23,315] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:23,536] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:23,686] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:23,922] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:24,071] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-05-16 06:45:24,221] WARN Bootstrap broker Node1:6667 disconnected (org.apache.kafka.clients.NetworkClient)

Why it happens, any other encoutered?

[#2]

@Zhao Chaofeng

Can you please share the exact command that you are using for consumer and producer?

Also do you see any issue when you append the following option to your Kafka Producer/Consumer?

--security-protocol SASL_PLAINTEXT

.

Also if you have set "security.inter.broker.protocol" then what is the value for this property?

.

[#3]

Hello, My produce command is

$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list node3:6667 --topic my-replicated-topic --producer.config /etc/kafka/conf/ssl/producer.properties

consume command is

./bin/kafka-console-consumer.sh --bootstrap-server node1:6667 --topic my-replicated-topic --new-consumer --consumer.config /etc/kafka/conf/ssl/producer.properties --from-beginning

I use SSL to authorize not Kerberos, so I add

--security-protol SSL

option to my Kafka Producer/Consumer. But in produce process, console prints messages as follows

[2017-05-16 16:03:21,447] WARN Error while fetching metadata with correlation id 0 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:21,548] WARN Error while fetching metadata with correlation id 1 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:21,655] WARN Error while fetching metadata with correlation id 2 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:21,765] WARN Error while fetching metadata with correlation id 3 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:21,874] WARN Error while fetching metadata with correlation id 4 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:21,984] WARN Error while fetching metadata with correlation id 5 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,093] WARN Error while fetching metadata with correlation id 6 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,202] WARN Error while fetching metadata with correlation id 7 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,313] WARN Error while fetching metadata with correlation id 8 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,423] WARN Error while fetching metadata with correlation id 9 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,532] WARN Error while fetching metadata with correlation id 10 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,642] WARN Error while fetching metadata with correlation id 11 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,751] WARN Error while fetching metadata with correlation id 12 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,858] WARN Error while fetching metadata with correlation id 13 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:22,967] WARN Error while fetching metadata with correlation id 14 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:23,076] WARN Error while fetching metadata with correlation id 15 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:23,183] WARN Error while fetching metadata with correlation id 16 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:23,292] WARN Error while fetching metadata with correlation id 17 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:23,400] WARN Error while fetching metadata with correlation id 18 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:23,512] WARN Error while fetching metadata with correlation id 19 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:23,629] WARN Error while fetching metadata with correlation id 20 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:23,738] WARN Error while fetching metadata with correlation id 21 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-05-16 16:03:23,848] WARN Error while fetching metadata with correlation id 22 : {my-replicated-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

[#4]

Yes, I set "security.inter.broker.protocol" as "SSL".

[#5]

My kafka server.properties content is

# Generated by Apache Ambari. Tue May 16 06:15:33 2017
    
advertised.host.name=172.21.9.35
advertised.listeners=PLAINTEXT://172.21.9.35:6666,SSL://172.21.9.35:6667
allow.everyone.if.no.acl.found=true
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
auto.create.topics.enable=true
auto.leader.rebalance.enable=true
compression.type=producer
controlled.shutdown.enable=true
controlled.shutdown.max.retries=3
controlled.shutdown.retry.backoff.ms=5000
controller.message.queue.size=10
controller.socket.timeout.ms=30000
default.replication.factor=1
delete.topic.enable=false
external.kafka.metrics.exclude.prefix=kafka.network.RequestMetrics,kafka.server.DelayedOperationPurgatory,kafka.server.BrokerTopicMetrics.BytesRejectedPerSec
external.kafka.metrics.include.prefix=kafka.network.RequestMetrics.ResponseQueueTimeMs.request.OffsetCommit.98percentile,kafka.network.RequestMetrics.ResponseQueueTimeMs.request.Offsets.95percentile,kafka.network.RequestMetrics.ResponseSendTimeMs.request.Fetch.95percentile,kafka.network.RequestMetrics.RequestsPerSec.request
fetch.purgatory.purge.interval.requests=10000
host.name=172.21.9.35
kafka.ganglia.metrics.group=kafka
kafka.ganglia.metrics.host=localhost
kafka.ganglia.metrics.port=8671
kafka.ganglia.metrics.reporter.enabled=true
kafka.metrics.reporters=org.apache.hadoop.metrics2.sink.kafka.KafkaTimelineMetricsReporter
kafka.timeline.metrics.host=node3
kafka.timeline.metrics.maxRowCacheSize=10000
kafka.timeline.metrics.port=6188
kafka.timeline.metrics.protocol=http
kafka.timeline.metrics.reporter.enabled=true
kafka.timeline.metrics.reporter.sendInterval=5900
kafka.timeline.metrics.truststore.password=bigdata
kafka.timeline.metrics.truststore.path=/etc/security/clientKeys/all.jks
kafka.timeline.metrics.truststore.type=jks
leader.imbalance.check.interval.seconds=300
leader.imbalance.per.broker.percentage=10
listeners=PLAINTEXT://node1:6666,SSL://node1:6667
log.cleanup.interval.mins=10
log.dirs=/kafka-logs
log.index.interval.bytes=4096
log.index.size.max.bytes=10485760
log.retention.bytes=-1
log.retention.hours=168
log.roll.hours=168
log.segment.bytes=1073741824
message.max.bytes=1000000
min.insync.replicas=1
num.io.threads=8
num.network.threads=3
num.partitions=1
num.recovery.threads.per.data.dir=1
num.replica.fetchers=1
offset.metadata.max.bytes=4096
offsets.commit.required.acks=-1
offsets.commit.timeout.ms=5000
offsets.load.buffer.size=5242880
offsets.retention.check.interval.ms=600000
offsets.retention.minutes=86400000
offsets.topic.compression.codec=0
offsets.topic.num.partitions=50
offsets.topic.replication.factor=3
offsets.topic.segment.bytes=104857600
port=6667
principal.builder.class=org.apache.kafka.common.security.auth.DefaultPrincipalBuilder
producer.purgatory.purge.interval.requests=10000
queued.max.requests=500
replica.fetch.max.bytes=1048576
replica.fetch.min.bytes=1
replica.fetch.wait.max.ms=500
replica.high.watermark.checkpoint.interval.ms=5000
replica.lag.max.messages=4000
replica.lag.time.max.ms=10000
replica.socket.receive.buffer.bytes=65536
replica.socket.timeout.ms=30000
security.inter.broker.protocol=SSL
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
socket.send.buffer.bytes=102400
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.key.password=hadoop
ssl.keystore.location=/etc/kafka/conf/ssl/kafka.server.keystore.jks
ssl.keystore.password=hadoop
ssl.keystore.type=JKS
ssl.truststore.location=/etc/kafka/conf/ssl/kafka.server.truststore.jks
ssl.truststore.password=hadoop
ssl.truststore.type=JKS
super.users=User:CN=node1,OU=test,O=test,L=test,ST=test,C=te
zookeeper.connect=node2:2181,node3:2181,node4:2181
zookeeper.connection.timeout.ms=25000
zookeeper.session.timeout.ms=30000
zookeeper.sync.time.ms=2000

[#6]

Hi, were you able to solve this issue? I am having the same problem.

[#7]

@Zhao Chaofeng

- were you able to resolve this issue ? I'm getting the same error..

[#8]

I had the same issue....it seems that the consumer/producer scripts are not taking in consideration the security.protocol setting used in the config file for consumer/producer.

By using the following option everything works as expected.

--security-protocol SSL

Let me know if everything works for you guys, @Saulo Sobreiro, @Zhao Chaofeng

[#9]

Yes, I add it in my execute command, but that does't work.