I have been trying to deploy Kafka using Helm charts. So I defined NodePort service for Kafka pods. I checked console Kafka producer and consumer with the same hosts and ports - they work properly. However, when I create Spark application as data consumer and Kafka as producer they are not able to connect to the Kafka service0. I used minikube ip (instead of node ip) for the host and service NodePort port. Although, in Spark logs, I saw that NodePort service resolves endpoints and brokers are discovered as pods addressed and ports:
INFO AbstractCoordinator: [Consumer clientId=consumer-1, groupId=avro_data] Discovered group coordinator 172.17.0.20:9092 (id: 2147483645 rack: null)
INFO ConsumerCoordinator: [Consumer clientId=consumer-1, groupId=avro_data] Revoking previously assigned partitions []
INFO AbstractCoordinator: [Consumer clientId=consumer-1, groupId=avro_data] (Re-)joining group
WARN NetworkClient: [Consumer clientId=consumer-1, groupId=avro_data] Connection to node 2147483645 (/172.17.0.20:9092) could not be established. Broker may not be available.
INFO AbstractCoordinator: [Consumer clientId=consumer-1, groupId=avro_data] Group coordinator 172.17.0.20:9092 (id: 2147483645 rack: null) is unavailable or invalid, will attempt rediscovery
WARN NetworkClient: [Consumer clientId=consumer-1, groupId=avro_data] Connection to node 2 (/172.17.0.20:9092) could not be established. Broker may not be available.
WARN NetworkClient: [Consumer clientId=consumer-1, groupId=avro_data] Connection to node 0 (/172.17.0.12:9092) could not be established. Broker may not be available.
How this behavior can be changed?
NodePort service definition looks like this:
kind: Service
apiVersion: v1
metadata:
name: kafka-service
spec:
selector:
app: cp-kafka
release: my-confluent-oss
ports:
- protocol: TCP
targetPort: 9092
port: 32400
nodePort: 32400
type: NodePort
Spark consumer configuration:
def kafkaParams() = Map[String, Object](
"bootstrap.servers" -> "192.168.99.100:32400",
"schema.registry.url" -> "http://192.168.99.100:8081",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[KafkaAvroDeserializer],
"group.id" -> "avro_data",
"auto.offset.reset" -> "earliest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)
Kafka producer configuration:
props.put("bootstrap.servers", "192.168.99.100:32400")
props.put("client.id", "avro_data")
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer")
props.put("schema.registry.url", "http://192.168.99.100:32500")
All the K8s services for Kafka:
NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE
kafka-service NodePort 10.99.113.234 <none> 32400:32400/TCP 6m34s
kubernetes ClusterIP 10.96.0.1 <none> 443/TCP 27d
my-confluent-oss-cp-kafka ClusterIP 10.100.156.108 <none> 9092/TCP 102m
my-confluent-oss-cp-kafka-connect ClusterIP 10.99.78.89 <none> 8083/TCP 102m
my-confluent-oss-cp-kafka-headless ClusterIP None <none> 9092/TCP 102m
my-confluent-oss-cp-kafka-rest ClusterIP 10.100.152.109 <none> 8082/TCP 102m
my-confluent-oss-cp-ksql-server ClusterIP 10.96.249.202 <none> 8088/TCP 102m
my-confluent-oss-cp-schema-registry ClusterIP 10.109.27.45 <none> 8081/TCP 102m
my-confluent-oss-cp-zookeeper ClusterIP 10.102.182.90 <none> 2181/TCP 102m
my-confluent-oss-cp-zookeeper-headless ClusterIP None <none> 2888/TCP,3888/TCP 102m
schema-registry-service NodePort 10.103.100.64 <none> 32500:32500/TCP 33m
zookeeper-np NodePort 10.98.180.130 <none> 32181:32181/TCP 53m
I had a similar issue when I was trying to access kafka broker (cp-helm-chart) running on minikube from outside.
Here how I resolved it. Before you install using helm install from local repository.
Now you can access the kafka broker running within k8s cluster from outside by pointing the bootstrap.servers to 196.169.99.100:31090