Using the Apache Kafka Java client (0.9), I'm trying to send a long series of records to the broker using the Kafka Producer class.
The asynchronous send method returns immediately for a while, then starts blocking on each call for a short time period. After around thirty seconds, the client starts throwing exceptions (TimeoutException), with the message "Batch expired".
What circumstances cause this exception to be thrown?
This exception indicates you are queueing records at a faster rate than they can be sent.
When you call the send method, the ProducerRecord will be stored in an internal buffer for sending to the broker. The method returns immediately once the ProducerRecord has been buffered, regardless of whether it has been sent.
Records are grouped into batches for sending to the broker, to reduce the transport overheard per message and increase throughput.
Once a record is added a batch, there is a time limit for sending that batch to ensure it has been sent within a specified duration. This is controlled by the Producer configuration parameter, request.timeout.ms, which defaults to thirty seconds.
If the batch has been queued longer than the timeout limit, the exception will be thrown. Records in that batch will be removed from the send queue.
Increasing the timeout limit, using the configuration parameter, will allow the client to queue batches for longer before expiring.
I got this exception in a completely different context.
I have setup a mini cluster of a zookeeper vm, a broker vm and a producer/consumer vm. I opened all neccessary ports on the server (9092) and on the zookeeper (2181) and then tried to publish a message from the consumer/publisher vm to the broker. I got the exception mentioned by the OP, but since I had only published one single message so far (or at least I tried to), the solution couldn't be to increase the timeout or batch size. So I searched on and found this mailing list describing a similar problem I had when trying to consume messages from within the consumer/producer vm (ClosedChannelException): http://grokbase.com/t/kafka/users/152jsjekrm/having-trouble-with-the-simplest-remote-kafka-config The last post in this mailing list actually describes how to solve the problem.
Long story short, if you face both the ChannelClosedException and the Batch Expired exception, you likely have to change this line to the following in the server.config file and restart the broker:
advertised.host.name=<broker public IP address>
If it isn't set, it falls back to the host.name property (which probably isn't set neither) and then falls back to the canonical host name of the InetAddress Java class, which finally isn't correct of course and thus confusing remote nodes.
I am using Kafka Java client version 0.11.0.0. I also started seeing the same pattern in failure to produce large messages consistently. It was passing for few of the messages, and failing for some others. (Although both passed and failed messages were of the same size).In my case, each message size was around 60KB, which is far higher than Kafka's default batch.size of 16kB, also my linger.ms was set to default of 0. This error is being thrown as the Producer client is timing out before it can receive a succesful response from the server.Basically, in my code , this call was timing out : kafkaProd.send(pr).get(). To fix this, I had to increase the Producer client's default request.timeout.ms to 60000
Had a similar issue with Kafka running in a docker-compose. My docker-compose.yml was set with
KAFKA_ADVERTISED_HOST_NAME: kafka
ports:
- 9092:9092
But when I tried to send a message with camel from outside docker
to("kafka:test?brokers=localhost:9092")
I got a TimeoutException. I solved it by adding
127.0.0.1 kafka
to Windows\System32\drivers\etc\hosts file and then changing my camel url to
to("kafka:test?brokers=kafka:9092")
i solved it.
My Kafka is deployed in a Docker container, and the container's network mode is the bridge,The host and container use port mappings,and i I changed the default port to 9102 for the Kafka server.
The configuration items in server.properties to solve the problem are these two: listeners advertised.listeners
I tried several combinations:
success：
listeners=PLAINTEXT://:9102
advertised.listeners=PLAINTEXT://192.168.0.136:9102
server can't start:
listeners=PLAINTEXT://192.168.0.136:9102
advertised.listeners=PLAINTEXT://192.168.0.136:9102
timeout error:
listeners=PLAINTEXT://:9102
advertised.listeners=PLAINTEXT://:9102
