CODEWITHSUNDEEP
kubernetesapache-kafkapykafka
GecKo
GecKo

Reputation: 173

Unable to connect kafka in kubernetes cluster from python-producer

I have a python fastapi app and kafka in one kubernetes cluster, but in different namespaces. Python app produces messages from client to kafka topic. Firstly, in python script I try to connect to kafka broker in following way:

producer = KafkaProducer(
    bootstrap_servers=os.environ["KAFKA_SERVER"],
    sasl_plain_username=os.environ["KAFKA_BROKER_USERNAME"],
    sasl_plain_password=os.environ["KAFKA_BROKER_PASSWORD"],
    security_protocol="PLAINTEXT",
    sasl_mechanism="PLAINTEXT",
    value_serializer=lambda v: v.encode('utf-8')
)

The value of the KAFKA_SERVER is the name of the kafka service in cluster. In this case is: gb-kafka.kafka.svc.gb.local:9092 When app starts it crushes nd raise the error:

Traceback (most recent call last):
  File "/code/main.py", line 45, in <module>
    producer = KafkaProducer(
  File "/usr/local/lib/python3.10/site-packages/kafka/producer/kafka.py", line 381, in __init__
    client = KafkaClient(metrics=self._metrics, metric_group_prefix='producer',
  File "/usr/local/lib/python3.10/site-packages/kafka/client_async.py", line 244, in __init__
    self.config['api_version'] = self.check_version(timeout=check_timeout)
  File "/usr/local/lib/python3.10/site-packages/kafka/client_async.py", line 927, in check_version
    raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable

I confugirate kafka with help of Helm, accurately with bitnami/kafka. I didn't do some massive changes in values.yaml file, just changed listeners configuration:

listeners: "PLAINTEXT://:9092"
advertisedListeners: "PLAINTEXT://gb-kafka.kafka.svc.gb.local:9092"
listenerSecurityProtocolMap: "PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
allowPlaintextListener: true
interBrokerListenerName: PLAINTEXT

kubectl get all command get the following result (ip addresses have changed):

NAME                       READY   STATUS    RESTARTS   AGE
pod/gb-kafka-0             1/1     Running   0          45m
pod/gb-kafka-zookeeper-0   1/1     Running   0          4d2h

NAME                                  TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)                      AGE
service/gb-kafka                      ClusterIP   10.10.10.543    <none>        9092/TCP                     4d2h
service/gb-kafka-headless             ClusterIP   None            <none>        9092/TCP,9093/TCP            4d2h
service/gb-kafka-zookeeper            ClusterIP   10.222.01.220   <none>        2181/TCP,2888/TCP,3888/TCP   4d2h
service/gb-kafka-zookeeper-headless   ClusterIP   None            <none>        2181/TCP,2888/TCP,3888/TCP   4d2h

NAME                                  READY   AGE
statefulset.apps/gb-kafka             1/1     4d2h
statefulset.apps/gb-kafka-zookeeper   1/1     4d2h

The most interesting thing, if I just running simple pod with python connection to to the kafka with the same configuration and simple producer without using fastapi it connetc to the broker well.

Upvotes: 0

Views: 409

Answers (1)

itpaman
itpaman

Reputation: 1

If you're using bitnami/kafka chart, You don't need to change listeners configuration when you connect from inside kubernetes cluster.

You just can access to kafka by kubernetes's internal dns policy.

try gb-kafka.kafka.svc.cluster.local:9092 or gb-kafka-headless.kafka.svc.cluster.local:9092

Upvotes: 0

Related Questions