CODEWITHSUNDEEP
dockerapache-kafkaapache-zookeeperkuberneteskafka-python
NegatioN
NegatioN

Reputation: 727

Creating topics, but getting Kafka FailedPayloadsError with Python on Kubernetes

I'm using a SimpleProducer in the python kafka-library. This script has worked flawlessly previously with other more hard-configured kafka-setups I've tried.

kafka = KafkaClient(u'[masterNodeIp]:[servicePort]')
producer = SimpleProducer(kafka)
#make a simple message, while true run
producer.send_messages(b'oneMoreTopic', sentence)

After running this script once, I get this response in the python-console.

kafka.common.LeaderNotAvailableError: TopicMetadata(topic='oneMoreTopic', error=5, partitions=[])

I can then go into my Node on my zookeeper.log and see:

2015-09-14 12:16:32,276 - INFO  [ProcessThread(sid:3  cport:-1)::PrepRequestProcessor@627] - Got user-level KeeperException when processing sessionid:0x34fcb982d030000 type:setData cxid:0x71 zxid:0x1000000d8 txntype:-1 reqpath:n/a Error Path:/config/topics/oneMoreTopic Error:KeeperErrorCode = NoNode for /config/topics/oneMoreTopic
2015-09-14 12:16:32,278 - INFO  [ProcessThread(sid:3 cport:-1)::PrepRequestProcessor@627] - Got user-level KeeperException when processing sessionid:0x34fcb982d030000 type:create cxid:0x72 zxid:0x1000000d9 txntype:-1 reqpath:n/a Error Path:/config/topics Error:KeeperErrorCode = NodeExists for /config/topics
2015-09-14 12:16:32,302 - INFO  [ProcessThread(sid:3 cport:-1)::PrepRequestProcessor@627] - Got user-level KeeperException when processing sessionid:0x34fcb982d030000 type:create cxid:0x7b zxid:0x1000000dc txntype:-1 reqpath:n/a Error Path:/brokers/topics/oneMoreTopic/partitions/0 Error:KeeperErrorCode = NoNode for /brokers/topics/oneMoreTopic/partitions/0
2015-09-14 12:16:32,304 - INFO  [ProcessThread(sid:3 cport:-1)::PrepRequestProcessor@627] - Got user-level KeeperException when processing sessionid:0x34fcb982d030000 type:create cxid:0x7c zxid:0x1000000dd txntype:-1 reqpath:n/a Error Path:/brokers/topics/oneMoreTopic/partitions Error:KeeperErrorCode = NoNode for /brokers/topics/oneMoreTopic/partitions

This seems to just be the Zookeeper creating a new Znode for the topic since it is not there from before. And the Kafka server.log prints:

[2015-09-14 12:16:32,282] INFO Topic creation {"version":1,"partitions":{"0":[10200119]}} (kafka.admin.AdminUtils$)
[2015-09-14 12:16:32,287] INFO [KafkaApi-10200219] Auto creation of topic oneMoreTopic with 1 partitions and replication factor 1 is successful! (kafka.server.KafkaApis)
[2015-09-14 12:16:51,579] INFO Closing socket connection to /10.240.1.94. (kafka.network.Processor)

However, my message is never posted to the topic and the next time i run the python-script i always get:

kafka.common.FailedPayloadsError

In the cases where i made it work, the advertised.host.name was always the external IP of the node, but I can't seem to get that working through Kubernetes. Would it be possible to make the external IP callable from the container parhaps?

my kafka/config/server.properties looks something like this for all brokers:

broker.id=10200121
host.name=kafka-f8p06
advertised.host.name=kafka-f8p06
++

Upvotes: 1

Views: 2002

Answers (2)

NegatioN
NegatioN

Reputation: 727

The root cause of my problem was fixed by making sure all my nodes could reach the internet by doing

sudo iptables -t nat -A POSTROUTING ! -d 10.0.0.0/8 -o ens4v1 -j MASQUERADE

After this, I changed the advertised.host.name to the external IP address I'm trying to reach my broker from. So if my GCE-node is located on 156.99.33.101, with a nodePort open on 33777 for my kafka-service, I would put:

advertised.host.name=156.99.33.101
advertised.host.port=33777

The advertised.host.name is what kafka uses to connect back to itself when one of the brokers get a request, so using the external address at least makes it reachable. Although I'm not sure if there are any consequenses to directing it outside the local address-space.

Topic about LeaderNotAvailableException remotely

Upvotes: 0

MrE
MrE

Reputation: 20808

broker.id=10200121
host.name=kafka-f8p06  <----- use IP here
advertised.host.name=kafka-f8p06  <---- use IP here

I think you should have IPs for host.name and advertised.host.name as K8s does not resolve Pods by hostname but it does by IP.

Your kafka nodes probably can't talk to each other that way and can't find the leader.

Upvotes: 1

Related Questions