Blizzard / Blizzard/node-rdkafka

Kafka consumer reads same message again after consumer group rebalances

Open
#1,079 1 comment 4 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.