Bug: #clearCacheAndResetPositions seeks to wrong offset causing message duplication

[pyh5431]

Description

There's a bug in _consumer.js where the #clearCacheAndResetPositions method seeks to the wrong offset, causing already-processed messages to be reprocessed.

Bug Location

File: consumer.js

Method: #clearCacheAndResetPositions

The comment on the method clearly states the intended behavior:

/* Seek to stored offset for each topic partition. It's possible that we've
 * consumed messages upto N from the internalClient, but the user has stale'd the cache
 * after consuming just k (< N) messages. We seek back to last consumed offset + 1. */

However, the actual implementation does not add +1 to the offset:

const lastConsumedOffsets = this.#lastConsumedOffsets.get(key);
const topicPartitionOffsets = [
  {
    topic: topicPartition.topic,
    partition: topicPartition.partition,
    offset: lastConsumedOffsets.offset,  // Bug: should be lastConsumedOffsets.offset + 1
    leaderEpoch: lastConsumedOffsets.leaderEpoch,
  }
];
seeks.push(this.#seekInternal(topicPartitionOffsets));

Expected Behavior

When cache expiration triggers #clearCacheAndResetPositions, the consumer should seek to lastConsumedOffset + 1 to avoid reprocessing the last successfully processed message.

Actual Behavior

The consumer seeks to lastConsumedOffset, causing the last successfully processed message to be consumed and processed again.

[PratRanj07]

Hey
Thanks for raising this issue! We have identified this as a bug.
Created internal ticket 4246.
We will be looking into this!

I have attached a test script to reproduce the bug
`const { Kafka } = require('./lib').KafkaJS;

const TOPIC = 'test-duplicate-bug-' + Date.now();
const GROUP_ID = 'test-group-' + Date.now();
const BROKER = process.env.KAFKA_BROKER || 'localhost:9092';

const consumedOffsets = [];

function sleep(ms) {
return new Promise(resolve => setTimeout(resolve, ms));
}

async function produce(kafka, count) {
const producer = kafka.producer();
await producer.connect();

const messages = Array.from({ length: count }, (_, i) => ({
key: key-${i},
value: message-${i},
partition: 0,
}));

await producer.send({ topic: TOPIC, messages });
await producer.disconnect();

console.log(Produced ${count} messages to ${TOPIC});
}

async function consume(kafka) {
const consumer = kafka.consumer({
'group.id': GROUP_ID,
'auto.offset.reset': 'earliest',
'enable.auto.commit': false,
'session.timeout.ms': 6000,
'max.poll.interval.ms': 10000, // Short timeout to trigger cache expiration
});

await consumer.connect();
await consumer.subscribe({ topics: [TOPIC] });

let count = 0;
let duplicates = 0;

return new Promise((resolve) => {
const timeout = setTimeout(async () => {
await consumer.disconnect();
resolve({ count, duplicates });
}, 30000);

consumer.run({
  eachMessage: async ({ message }) => {
    const offset = message.offset;
    const isDuplicate = consumedOffsets.includes(offset);
    
    if (isDuplicate) {
      duplicates++;
      console.log(`DUPLICATE: offset ${offset} already consumed`);
    } else {
      console.log(`Consumed: offset ${offset}`);
    }
    
    consumedOffsets.push(offset);
    count++;
    
    // After 3 messages, block longer than max.poll.interval.ms
    // to trigger cache expiration and the bug
    if (count === 3) {
      console.log(`\nBlocking for 12s to trigger cache expiration...`);
      console.log(`Last offset before block: ${offset}\n`);
      await sleep(12000);
    }
    
    if (duplicates > 0 || count >= 10) {
      clearTimeout(timeout);
      await consumer.disconnect();
      resolve({ count, duplicates });
    }
  },
});

});
}

async function main() {
console.log(Broker: ${BROKER});
console.log(Topic: ${TOPIC}\n);

const kafka = new Kafka({ 'bootstrap.servers': BROKER });

try {
await produce(kafka, 10);
const result = await consume(kafka);

console.log(`\n--- Results ---`);
console.log(`Total consumed: ${result.count}`);
console.log(`Duplicates: ${result.duplicates}`);

if (result.duplicates > 0) {
  console.log(`\nBug confirmed: #clearCacheAndResetPositions seeks to wrong offset`);
  console.log(`Offsets consumed: ${consumedOffsets.join(', ')}`);
}

} catch (err) {
console.error('Error:', err.message);
process.exit(1);
}
}

main();`

[pyh5431]

Hi, thanks for identifying this as a bug earlier.

I noticed that the latest release (v1.8.2) still seems to has this bug.
Are there any updates on the internal ticket 4246?

Thanks!