External service for kafka not working

[juanli2]

Hi,
I tried to use a NodePort type to expose the kafka service out.
Attached my service yml file.

And here's the service descrition:
$ kubectl describe svc kafka -n kafka
Name: kafka
Namespace: kafka
Labels:
Selector: app=kafka
Type: NodePort
IP: 100.73.225.207
Port: 9092/TCP
NodePort: 32093/TCP
Endpoints: 10.44.0.10:9092,10.44.0.11:9092,10.44.0.12:9092
Session Affinity: None

But when I tried using port 32039 external to connect to the kafka service it seems not working.

$ ./bin/kafka-console-producer.sh --broker-list ${master-ip}:32093 --topic test2
abc
[2016-11-18 15:26:58,157] ERROR Error when sending message to topic test2 with key: null, value: 3 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for test2-0 due to 1512 ms has passed since batch creation plus linger time

I'm pretty sure that the connection to the master-ip is working, and the port 32039 is listening in the cluster.

It works for the zookeeper, but I'm not sure why the kafka not working

[solsson]

The example kafka service 30service.yml doesn't use NodePort, so it will only be accessible within the cluster. Port will be 9092 but not exposed on nodes.

If you modified the service, what kubernetes environment do you use? Does it require a LoadBalancer type service? This repository has only focused on kafka access within the cluster.

Have you tried any of the pods in https://github.com/Yolean/kubernetes-kafka/tree/master/test ?

[juanli2]

The access within the pod is OK. Just I want to test expose the port out. It's in my local lab. Not on cloud. So I use NodePort Type. It works well for zookeeper NodePort. But it just failed when accessed outside.

What's your suggestion if I want to expose the 9092 port out.

[juanli2]

[2016-11-21 15:09:38,373] WARN [console-consumer-72367_JUANLI2-M-G38Q-1479712160584-889927b7-leader-finder-thread], Failed to find leader for Set(test1-0) (kafka.consumer.ConsumerFetcherManager$LeaderFinderThread)
kafka.common.KafkaException: fetching topic metadata for topics [Set(test1)] from broker [ArrayBuffer(BrokerEndPoint(1,kafka-1.broker.kafka.svc.cluster.local,9092), BrokerEndPoint(2,kafka-2.broker.kafka.svc.cluster.local,9092), BrokerEndPoint(0,kafka-0.broker.kafka.svc.cluster.local,9092))] failed
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:73)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:65)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
Caused by: java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
... 3 more

It seems I need a LoadBlancer to those brokers

[donstaheli]

A kafka client that supports broker redirection will work in this scenario, just map the three internal broker endpoints to the one used in your connection string.

[vhsanche]

Having the same issue. Were you able to figure out how to expose the kafka service @juanli2 using NodePort?

[solsson]

I'm not sure the kafka protocol allows such a thing. See an ongoing investigation in #21. I think consumers must be able to address each broker separately, though it can get the list of brokers from one or more "bootstrap" names. Kubernetes services are inherently round-robin.

[donstaheli]

I was able to get around this using broker redirection in no-kafka. What happens is that upon connection to a broker, it sends a list of known brokers which then replace whatever endpoint you used to connect with. These are the kubernetes cluster endpoints, however, which will not be the same as what the consumer is connecting with. The broker redirection feature lets you map the external endpoint with the internal one.

[vhsanche]

Ahh right on, let me try that, and will report back!

[solsson]

@donstaheli Have you tried that setup with a topic that has fewer replicas than you have brokers?

[donstaheli]

No, I haven't. What is the use case demanding that can't be met with partitioning?

[solsson]

It's more of a gotcha than a use case. In #21 we suspect that when consumers can't resolve individual brokers (internal or external services alike) they will not know how to locate a topic unless it lives on the broker they happen to hit.

I don't think a client lib can solve that, because k8s services are round-robin. There's a discussion about that in kubernetes/kubernetes#36415.

[vhsanche]

Sorry for the late update, but I looked at the broker redirection no-kafka and this is not what I'm looking for. So, my question is similar to @juanil2. Let me explain. I used the repo to create the zookeeper and Kafka statefulsets. I was successful creating everything as advertised. I am able to access Kafka by either hitting the broker-0.broker.... and so forth or by using the Kafka.default.svc.... I then went ahead and modified the 30service.yml with type: NodePort and attempt to hit this service externally to the k8 cluster, and I get the same error @juanil2 reported. If I understand nodeport services correctly, this should work? Btw, my k8 cluster is in not in a cloud environment. Any ideas ?

[solsson]

I think the no-kafka docs on remapping confirm my position that Kafka isn't meant to be used through a round-robin load balancer. It might work within a cluster though, where you can - depending on config - resolve the addresses to individual brokers that you get from your initial connection through the service.

[solsson]

You could try one service + NodePort per kafka broker, together with remapping.