confluentinc / confluentinc/librdkafka

How to properly reconsume messages after their processing failed in High-Level balanced consumer?

Open
#4,841 2 comments 0 reactions 0 assignees View on GitHub
Dominant language
C
Stars
1k
Forks
3.3k
Avg merge
5d 12h
Merged PRs (30d)
6

Description

### Discussed in https://github.com/confluentinc/librdkafka/discussions/4722

Originally posted by **fdr400** May 17, 2024
Hello everyone! I am currently implementing the driver for Kafka in @userver-framework. The code is available here: https://github.com/userver-framework/userver/tree/develop/kafka

It provides an interface for processing message batches received from the topics consumer has subscribed on. Common use-case of consumer looks like this:
```cpp
consumer.Start([](kafka::MessageBatchView batch) {
/* message batch processing */
consumer.AsyncCommit(); // wrapper for rd_kafka_commit(consumer_handle, 1)
});
```
`kafka::MessagesBatchView` is an alias for `std::span`, where `kafka::Message` is a C++ wrapper for `rd_kafka_message_t*`

The library periodically polls the message batches and invokes the user-provided callback passing the message batch view.

I am trying to support the message reconsumption when the callback throws an exception.
To implement such logic, current implementation closes the consumer and subscribes for the same topics, after catching the exception, to start reading the topic partition from the last committed offset.

But the implementation looks not optimal, because, for example, when using an EAGER rebalance strategy, after closing the consumer, all partitions are firstly revoked from all consumers in the current group and then assigned to them again.

I read lots of issues in the repo and googled a lot, but, unfortunately, did not find the, how to reread not committed messages
using the **balanced** consumer.

I saw several solutions to reread the messages, but, as I understand, the won't work when consumer **subscribed** to topics:
1. First solution is to fetch the current consumer assignment with `rd_kafka_assignment` and assign it to consumer with `rd_kafka_assign`. But in many issues I read that `rd_kafka_assign` assigns the static set to consumer and can not be used outside of rebalance_callback
2. Second, is to use the `rd_kafka_seek` to move the fetched offsets back to the commited values. But, again, I read that seek can used only after manual partition assignment with `rd_kafka_assign`
3. The last solution is to cache the unprocessed message batch to pass it again and again to the use callback, until the callback invokation succeeded. But in such solution consumer stops the polling loop and may be kicked from the consumer group after some time

So, the question is, how properly reconsume the uncommitted messages using the balanced consumer (`rd_kafka_subscribe`), with minimum overhead?

Contributor guide

Open the contributing guide

Research direction

No repository files or tests are named. Start by reading the API documentation and relevant implementation around rd_kafka_subscribe, rd_kafka_assignment, rd_kafka_assign, rd_kafka_seek, and rebalance_callback; done means documenting a supported way to reconsume uncommitted messages with a balanced consumer and its constraints.

Written by the indexing model from the issue text.

Assessment

Tech stack
c, cpp
Domain
distributed-systems
Issue type
Documentation
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.