Librdkafka: Commit Async causing request(s) timed out

[jleducq]

Description

I have some request timeout when using the manual async commit: Received an error event: server1:9092/1: 6521 request(s) timed out: disconnect(Local: Message timed out)

My application requires to commit transaction manually in order to ensure that the current transaction is successful before doing a commit. So the consumer currently takes the responsibility to do a async commit once the transaction is processed: consumer->commitAsync(message);

My scenario is the following:

• 1 Kafka topic with 5 partitions, 2 replicas
• 1 application which successfully produced 350k messages on each partition
• 1 application which consume from those 5 partitions
• The Kafka broker and the application share the same machine for now

Kafka consumer configuration:

• I have a rebalance_cb
• auto.commit.enable = false
• Default settings

The issue:
My application is able to consume around 300k messages at a good rate and the the following happens:

• My event callback receives the following at 11:51:37:
  11:51:37.650096135 [19235] ERROR - KafkaEventCallback: Received an error event: server1:9092/1: 2241 request(s) timed out: disconnect(Local: Message timed out)
  11:51:37.650113491 [19235] ERROR - KafkaEventCallback: Received an error event: server1:9092/1: 6521 request(s) timed out: disconnect(Local: Message timed out)
  11:51:37.660512925 [19235] ERROR - KafkaEventCallback: Received an error event: server1:9092/1: 10556 request(s) timed out: disconnect(Local: Message timed out)
• My rebalance callback receives a revoke partition the following at 11:51:37:
  11:51:37.660533148 [19235] INFO - KafkaRebalanceCallback: Rebalance of consumers: Local: Revoke partitions
• My event and consumer callback log the following at 11:52:35:
  11:52:35.658235508 [19234] INFO - KafkaEventCallback: Received a log event: server1:9092/1: 2811 request(s) timed out: disconnect(3, FAIL)
  11:52:35.658254799 [19235] ERROR - KafkaEventCallback: Received an error event: server1:9092/1: 2811 request(s) timed out: disconnect(Local: Message timed out)
  11:52:35.658270865 [19231] INFO - KafkaEventCallback: Received a log event: server1:9092/1: Metadata request failed: Local: Message timed out(4, METADATA)
  11:52:35.661602060 [19235] ERROR - KafkaConsumerProxy: Consumer failed to read message: Local: Message timed out
  11:52:37.658101566 [19234] INFO - KafkaEventCallback: Received a log event: server1:9092/1: 3017 request(s) timed out: disconnect(3, FAIL)
  11:52:37.658119054 [19235] ERROR - KafkaEventCallback: Received an error event: server1:9092/1: 3017 request(s) timed out: disconnect(Local: Message timed out)
• My broker logs the following at 11:52:38:
• My rebalance callback receives a assign partitions event at 11:52:38:
  11:52:38.308884530 [19235] INFO - KafkaRebalanceCallback: Rebalance of consumers: Local: Assign partitions
  11:52:38.308887115 [19235] INFO - KafkaRebalanceCallback: 5 partitions will be assigned to this APP consumer
  11:52:38.409012346 [19235] INFO - KafkaRebalanceCallback: Assigning topic: MY.TOPIC, Partition: 0
  11:52:38.409015587 [19235] INFO - KafkaRebalanceCallback: Assigning topic: MY.TOPIC, Partition: 1
  11:52:38.409016483 [19235] INFO - KafkaRebalanceCallback: Assigning topic: MY.TOPIC, Partition: 2
  11:52:38.409017055 [19235] INFO - KafkaRebalanceCallback: Assigning topic: MY.TOPIC, Partition: 3
  11:52:38.409017450 [19235] INFO - KafkaRebalanceCallback: Assigning topic: MY.TOPIC, Partition: 4

This result in a delay of 1 minute: no messages were consumed between 11:51:37 and 11:52:38. This problem happens again after another ~300k messages consumed. I also noticed the first timeout there is elapse time of ~5-10s where not messages are processed but no error is logged and no rebalance are triggered.
This error doesn't happen if I turn the auto commit to true and remove the commitAsync (no timeout, not rebalance), so this issue seems to be linked to the async commit.

Note that the issue happens each time I run the above scenario

How to reproduce

See my scenario above

Checklist

Please provide the following information:

• [ X ] librdkafka version (release number or git tag): 0.9.1
• [ X ] Apache Kafka version: 0.10
• [ X ] librdkafka client configuration: see above
• [ X ] Operating system: Linux centos 6.6
• [ X ] Using the legacy Consumer: No
• [ X ] Using the high-level KafkaConsumer: Yes
• [ X ] Provide logs (with debug=.. as necessary) from librdkafka
• [ X ] Provide broker log excerpts
• [ X ] Critical issue

[edenhill]

That looks nasty, can you try reproducing this issue with debug=broker,protocol enabled?

[jleducq]

I just reproduced the issue with debug logs. I can see a lot of those lines:

17:52:23.686339468 [2311] INFO - KafkaEventCallback: Received a log event: server1:9092/1: OffsetCommitRequest failed: Local: Timed out: explicit actions 0x0(7, REQERR)

Please see the attached file for the full logs.
kafkalog.txt

[edenhill]

If you are asynchronously committing offsets for each processed message, and the OffsetCommit request-response times are non-negligent, you will quite quickly build up a queue of OffsetCommitRequests waiting in-queue or in-flight to be processed and responded to by the broker.
When this queue grows large enough the latency to make it through is higher than the request timeout (socket.timeout.ms, def 60s) and these requests will start failing with "Local: Timed out", as you are seeing.

When more than socket.max.fails requests have timed out the broker connection is torn down and re-established and all this starts over.

The underlying problem is that your consumtion and processing rate is higher than the commit rate, which inevitably leads to these queue build ups.

...:

• Do you need to commit per message? You are using async commits so if the commit would fail for whatever reason (such as timeout on queue!) that commit is still lost. If you really need per-message commits you probably want sync commits as well, but that will really slow things down. Might be worth a shot though.
• What about not committing every message, but every say 100? or 1000? Whatever it takes to make the broker keep up with the commits.
• There are actually two layers of offset commit, first there is the offset_store() that stores the offset to be commited, then there's the actual commit which commits the stored offset. You could probably get away with store():ing every message's offset after processing but then relying on the auto offset committer to actually commit the offsets for you. Use enable.auto.offset.store=false to disable automatic offset store (which is typically done prior to your processing).

[jleducq]

Thanks for your answer.

I actually quite like the third solution using offset_store(). However according to the documentation this method should only be used with the old consumer. I am currently using the CPP api with the high-level consumer: KafkaConsumer. Is there a way of using it with the new consumer?

Note that I am also using virtual ErrorCode subscribe (const std::vector<std::string> &topics) = 0; and virtual ErrorCode assign (const std::vector<TopicPartition*> &partitions) = 0 so I do not have any Topic handles.

I do not need to commit on every message, I just don't want the following scenario to happen:

• Application consumes the message
• Application does internal processing
• Message offset is auto committed
• Application continues internal processing
• Application fails to finish the processing due to technical issue (e.g. database is down)

The idea was to commit once we are sure the message was processed correctly.

Is it the call to consume(1000) that will trigger an auto commit of previous processed offsets? Or, does the auto commit is handled by a different thread and can be triggered anytime?

[edenhill]

Ah yeah, offset_store() should be exposed on the KafkaConsumer class as well, without the Topic object requirement.. I'll fix that.

If enable.auto.commit is enabled the automatic commit is done every auto.commit.interval.ms by one of librdkafka's internal threads. The commit will commit the last stored offset for each partition.

[edenhill]

Can you give the offsets_store_826 branch a try, it has the new offsets_store() API that takes a list of partitions to store.

[jleducq]

I just tried it and it worked. Thanks for this! Is this feature will be added to the future 0.9.2 release?

It would be good to pass directly the message to the store method (like it is done for commit methods). I think that most user will call the function after processing a RdKafka::Message

For now I am doing an ugly workaround:

RdKafka::TopicPartition * topar = RdKafka::TopicPartition::create(message->topic_name(), message->partition());
topar->set_offset(message->offset());
std::vector<RdKafka::TopicPartition *> topars = {topar};
RdKafka::ErrorCode err = m_consumer->offsets_store(topars);

[jleducq]

Hi,
I see that the issue is mark as closed. Is there any plan to merge this enhancement with the master branch?
Thanks.

[edenhill]

You are right, seems like I missed to merge this.

[jleducq]

@edenhill Would it be possible to merge this branch in order to have the feature in the 0.9.4 release of librdkafka?

[edenhill]

Argh, missed it twice!
I'm sorry but it is too late to merge this for 0.9.4 now :(
I screwed up.

[edenhill]

Merge it

[jleducq]

@edenhill I see that the fix is on Master but I don't see it in the RC2 of 0.9.5. Would it be possible to add it in the final release? Thanks