Blizzard / Blizzard/node-rdkafka
Kafka consumer reads same message again after consumer group rebalances
- Dominant language
- JavaScript
- Stars
- 2.2k
- Forks
- 403
- PR merge metrics
- No merged PRs in 30d
Description
**Environment Information**
- OS [e.g. Mac, Arch, Windows 10]: Mac
- Node Version [e.g. 8.2.1]: 14.17.5
- NPM Version [e.g. 5.4.2]: 6.4.1
- C++ Toolchain [e.g. Visual Studio, llvm, g++]:
- node-rdkafka version [e.g. 2.3.3]: 2.18.0
**Steps to Reproduce**
I have two consumers running in my application both running on the same consumer group ID G1.
consumer1 is reading from topic1 and consumer2 is reading from topic2. autoCommit is disabled. I am manually committing each offset after message processing.
Both topics have only one partition.
MAX_MESSAGE_POLL is set to 1. So I am polling only 1 message at a time from the broker.
What I am noticing is, that if consumer1 is in middle of message processing and in the meantime consumer2 comes up. Consumer1 will re-read the same message again. I enabled rebalance_cb in kafka consumer config and got to know that broker rebalances the consumer group whenever a new consumer joins the group. It revokes and re-assigns the same partition again (since there is only one partition)
- Question 1: Why is consumer1 reading the same message again even after successfully committing the current offset?
Another thing I noticed is that if I provide a default implementation of rebalance_cb from https://www.npmjs.com/package/node-rdkafka#rebalancing
That is the below code:
```
'rebalance_cb': (err, assignment) => {
if (err.code === Kafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) {
// Note: this can throw when you are disconnected. Take care and wrap it in
// a try catch if that matters to you
this.consumer.assign(assignment);
} else if (err.code == Kafka.CODES.ERRORS.ERR__REVOKE_PARTITIONS){
// Same as above
this.consumer.unassign();
} else {
// We had a real error
console.error(err);
}
}
```
In this case, message is NOT read twice.
- Question 2: What is causing consumer1 to NOT read the same message again when providing a default implementation of rebalance_cb ? I mean without this function also, node-rdkafka must be following same strategy.
My expectation is that after a successful commit, same message should never be read again from same consumer group.
**node-rdkafka Configuration Settings**
```
brokerConfig: {
'metadata.broker.list': 'localhost:9092',
'group.id': `group-name`,
'event_cb': true,
'compression.codec': 'snappy',
'socket.keepalive.enable': true,
'enable.auto.commit': false,
'heartbeat.interval.ms': 250,
'queued.min.messages': 100,
'fetch.error.backoff.ms': 250,
'queued.max.messages.kbytes': 50,
},
topicConfig: {
'auto.offset.reset': 'earliest',
'request.required.acks': 1
}
```
Contributor guide
Research direction
Start by reproducing the two-consumer setup described in the issue with manual commits and one partition per topic. Compare behavior with and without the rebalance_cb implementation, focusing on partition revoke/assignment and the committed offset. Done means the duplicate-read behavior and callback effect are explained, or a focused fix is accompanied by a regression test.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- javascript, kafka, nodejs
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 30/100