Problem of Creating Topics in Kafka with Kerberos

[#1]

Hi,

After enabled Kerberos using Ambari, I got problem creating topics in Kafka using the kafka-topics.sh script. The topic was created, but its status is wrong without leader. It seems the topic is created with PLAINTEXT, while there is only PLAINTEXTSASL broker in the cluster after enabled Kerberos. The only configuration change I made is to chagne broker listener from 'PLAINTEXT://localhost:6667' to 'PLAINTEXTSASL://localhost:6667'. As posted in this question, I also changed the kafka-topics.sh to make it work with Kerberos. I am using HDP2.5.3.

$ ./kafka-topics.sh --zookeeper ip-10-0-0-149.ap-northeast-1.compute.internal --create --partitions 1 --replication-factor 1 --topic mytopic
Created topic "mytopic".

$ ./kafka-topics.sh --zookeeper ip-10-0-0-149.ap-northeast-1.compute.internal --describe --topic mytopic
Topic:mytopic	PartitionCount:1	ReplicationFactor:1	Configs:
	Topic: mytopic	Partition: 0	Leader: none	Replicas: 1001	Isr:

[#2]

@yjiang Could you please tell, as which user did you create the topic? Also could you please provide your server.properties file?

[#3]

I created the topic using a normal user (not the 'kafka' service user). Do I need to use 'kafka' user to create the topic?

Below is my server.properties.

advertised.listeners=PLAINTEXTSASL://ip-10-0-0-149.ap-northeast-1.compute.internal:6667
authorizer.class.name=org.apache.ranger.authorization.kafka.authorizer.RangerKafkaAuthorizer
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
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=ip-10-0-0-229.ap-northeast-1.compute.internal
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=PLAINTEXTSASL://ip-10-0-0-149.ap-northeast-1.compute.internal: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.to.local.class=kafka.security.auth.KerberosPrincipalToLocal
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=PLAINTEXTSASL
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
socket.send.buffer.bytes=102400
super.users=User:kafka
zookeeper.connect=ip-10-0-0-149.ap-northeast-1.compute.internal:2181
zookeeper.connection.timeout.ms=25000
zookeeper.session.timeout.ms=30000
zookeeper.set.acl=true
zookeeper.sync.time.ms=2000

[#4 · Accepted solution]

When you use a script, command, or API to create a topic, an entry is created under ZooKeeper. The only user with access to ZooKeeper is the service account running Kafka (by default, kafka). Therefore, the first step toward creating a Kafka topic on a secure cluster is to run kinit, specifying the Kafka service keytab. The second step is to create the topic.

1. Run kinit, specifying the Kafka service keytab. For example:

   kinit -k -t /etc/security/keytabs/kafka.service.keytab kafka/c6401.ambari.apache.org@EXAMPLE.COM

2. Next, create the topic. Run the kafka-topics.sh command-line tool with the following options:

   /bin/kafka-topics.sh --zookeeper <hostname>:<port> --create --topic <topic-name> --partitions <number-of-partitions> --replication-factor <number-of-replicating-servers>

   For example:

   /bin/kafka-topics.sh --zookeeper c6401.ambari.apache.org:2181 --create --topic test_topic --partitions 2 --replication-factor 2
   
   Created topic "test_topic".

[#5]

I created the topic using a normal user (not the 'kafka' service user). Do I need to use 'kafka' user to create the topic?

[#6]

yes as the kafka user

[#7]

@yjiang

To answer your question -

---------------

When you use a script, command, or API to create a topic, an entry is created under ZooKeeper. The only user with access to ZooKeeper is the service account running Kafka (by default, kafka). Therefore, the first step toward creating a Kafka topic on a secure cluster is to runkinit, specifying the Kafka service keytab. The second step is to create the topic.

----------------

See more in here: http://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.4.3/bk_secure-kafka-ambari/content/ch_secure-kaf...

I hope this helps.

[#8]

@yjiang Yes, please try creating the topic as kafka user because when we use kafka-topics.sh to create a test topic, what this script does is to create a node in zookeeper path - /broker/topics/test

then brokers thread gets notified that a new node is created and broker then creates actual data for topic test that is the metadata and physical data. But notice that Brokers are kafka/host@REALM, so if a user other than kafka creates a topic it gets permission for example world:anyone:r sasl:xyz:crdwa

So the new node that is created in zookeeper path will have these permissions. Now when the broker gets alerted and tries to create metadata and physical data for this new topic, it wont be able to because broker principal is kafka but topic's is xyz

[#9]

@yjiang did it work, when created as kafka user?

[#10]

@dbains, @mthiele, @Daniel Kozlowski,

Thank you. It works when created as kafka user.

To summarize all the steps -- from Ambari configs, to creating topic, granting permission and testing on kafka console producer/consumer scripts, I created this article Step by Step Recipe for Securing Kafka with Kerberos. Hope it saves others' time 🙂