[KAFKA-3686] Kafka producer is not fault tolerant

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 0.9.0.1
• Component/s: producer
• Assignee: Unassigned
• Reporter: Luca Bruno
• Created: 10/May/16 16:39
• Updated: 31/Jan/20 18:59

Description

Setup

I have a cluster of 3 kafka server, a topic with 12 partitions with replica 2, and a zookeeper cluster of 3 nodes.

Producer config:

 props.put("bootstrap.servers", "k1:9092,k2:9092,k3:9092");
 props.put("acks", "1");
 props.put("batch.size", 16384);
 props.put("retries", 3);
 props.put("buffer.memory", 33554432);
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer code:

 Producer<String, String> producer = new KafkaProducer<>(props);
 for(int i = 0; i < 10; i++) {
     Future<RecordMetadata> f = producer.send(new ProducerRecord<String, String>("topic", null, Integer.toString(i)));
     f.get();
 }

Problem

Cut the network between the producer (p1) and one of the kafka servers (say k1).

The cluster is healthy, hence the kafka bootstrap tells the producer that there are 3 kafka servers (as I understood it), and the leaders of the partitions of the topic.

So the producer will send messages to all of the 3 leaders for each partition. If the leader happens to be k1 for a message, the producer raises the following exception after request.timeout.ms:

Exception in thread "main" java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Batch Expired
	at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.valueOrError(FutureRecordMetadata.java:56)
	at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:43)
	at org.apache.kafka.clients.producer.internals.FutureRecordMetadata.get(FutureRecordMetadata.java:25)
	at Test.main(Test.java:25)
Caused by: org.apache.kafka.common.errors.TimeoutException: Batch Expired

In theory, the application should handle the failure. In practice, messages are getting lost, even though there are other 2 leaders available for writing.
I tried with values of acks both 1 and -1.

What I expected

Given the client library is automatically deciding the hashing / round robin schema for the partition, I would say it's not very important which partition the message is being sent to.
I expect the client library to handle the failure by sending the message to a partition of a different leader.

Neither kafka-clients nor rdkafka handle this failure. Given those are the main client libraries being used for kafka as far as I know, I find it a serious problem in terms of fault tolerance.

EDIT: I cannot add comments to this issue, don't understand why. To answer fpj yes, I want the first. In the case of network partitions I want to ensure my messages are stored. If the libraries don't do that, it means I have to reimplement them. Or otherwise, postpone sending such messages until the network partition resolves (which means implementing some kind of backlog on disk of the producer, which should instead be the kafka purpose after all). In both cases, it's something that is not documented and it's very inconvenient.