KafkaTimeoutError:

My code:

def kafka_producer():
     producer = KafkaProducer(bootstrap_servers='{0}:{1}'.format(HOST, PORT),
                         api_version=(0, 10, 1)
                         )
   try:
       future = producer.send('test', b'hello kafka')
       producer.flush()
       record_metadata = future.get(timeout=10)
       print(record_metadata)
except KafkaError as e:
       print(e)

My Kafka Broker (v 2.4.0) is running in OpenShift container. My Python producer code is similar to the above and that is causing the below issue.

KafkaTimeoutError: Batch for TopicPartition(topic='test', partition=0) containing 1 record(s) expired: 31 seconds have passed since batch creation plus linger time
   File "C:\Users\XXXXXXX\Anaconda3\lib\site-packages\kafka\producer\future.py", line 65, in get
    raise self.exception # pylint: disable-msg=raising-bad-type

My kafka-python version is 2.0.1

I am fighting on the issue for last two days. I am frustrated. Please assist how do you resolve it / why does it occur?

I have tested using the Kafka sh script to send the topic it’s working. So where is the problem in python?

Issue Analytics

State:
Created 3 years ago
Comments:7

Top GitHub Comments

1reaction

skskcco2o17commented, May 27, 2020

Hi tvoinarovskyi ,
Your understanding is absolutely correct. I run a openshift port forward command on my local windows machine to get access to Kafka Broker hosted in the OpenShift POD container and connect to it using bootstrap_servers=“localhost:9092”.

I have provided entry in /etc/hosts of Pod container (where Kafka Broker is running), NOT in my local machine. I open the pod container SSH terminal , go to /etc/hosts and do the changes , like below: 172.30.66.31 kafka-zookeeper-7-wfx9x

‘kafka-zookeeper-7-wfx9x’ is the host name of the OpenShift Pod where the Kafka broker container is running. I changed in local C:\Windows\System32\drivers\etc\hosts, as you suggested, as below and it worked : 127.0.0.1 kafka-zookeeper-7-wfx9x

But my bootstrap_servers=“localhost:9092” remains same.

I am happy, that it works…You rock, THANKS A LOT !!! You save my life. Please close this issue. 😃

0reactions

tvoinarovskyicommented, May 27, 2020

Glad it helped, a lot of people have problems with it. Kafka clients get a list of brokers on first connect from a random broker. That list shows hosts that Kafka broker think is correct, but in case of port forwarding is mostly incorrect (eg. even thou you connect using localhost:9092 it will still ask Kafka what is its hostname and connect using that one).

Top Results From Across the Web

How to solve kafka.errors.KafkaTimeoutError ... - Stack Overflow

When your code runs in Docker, localhost means the Docker container itself. This is why your connection times out, since Kafka isn't running ......

('Failed to update metadata after %s secs.', 60.0) · Issue #607 ...

kafka.errors.KafkaTimeoutError: KafkaTimeoutError: Failed to update metadata after 60.0 secs. My producer is running inside a docker container, but both kafka ...

KafkaTimeoutError: Failed to update metadata after 60.0 secs.

Hi guys, I'm data science student and i'm trying to use kafka (producer and consumer) download and handle tweets from twitter.

Python kafka.errors.KafkaTimeoutError() Examples

This page shows Python examples of kafka.errors.KafkaTimeoutError.

kafka.producer.kafka — kafka-python 2.0.2-dev documentation

Returns: FutureRecordMetadata: resolves to RecordMetadata Raises: KafkaTimeoutError: if unable to fetch topic metadata, or unable to obtain memory buffer ...