confluentinc / confluentinc/confluent-kafka-javascript
KafkaJS compat: pause() on an eachBatch consumer permanently loses prefetched messages (consumption wedges)
- Dominant language
- TypeScript
- Stars
- 304
- Forks
- 45
- Avg merge
- 11h 47m
- Merged PRs (30d)
- 5
Description
**Environment:** confluent-kafka-javascript 1.10.0 (also present on current master — `lib/` is identical), librdkafka 2.15.0, Node 22, brokers tested: Apache Kafka 3.9 (ZooKeeper and KRaft) and 4.0 (KRaft) — broker-independent.
## Summary
For a KafkaJS-compat consumer running `eachBatch`, calling `consumer.pause()` (from inside the callback or outside) permanently skips every message that had been prefetched into the compat layer's message cache beyond the batch being processed. Consumption appears wedged: after `resume()`, `consume()` fetches from the end of the already-prefetched range, the skipped messages are never redelivered, and their offsets are eventually committed. `disconnect()` on such a consumer can also hang.
The `eachMessage` path is unaffected, and small-volume pause tests (like the existing 4-message specs) don't trigger it — the cache must be warm enough that batches beyond the current one are already prefetched, which is exactly the situation for any real consumer using pause for flow control (the documented kafkajs idiom).
## Root cause
`#pauseInternal` discards prefetched messages from the message cache (`markStale`) and then repairs the fetch position by seeking to `#lastConsumedOffsets + 1`. But `#eachBatchPayload_resolveOffsets` — despite its own doc comment saying it stores "into the lastConsumedOffsets map that we use for seeking to the last consumed offset when forced to clear cache" — **never writes to that map**. Only `#messageProcessor` (the eachMessage path) does. So for eachBatch consumers the map is empty, no repair seek is issued, and the discarded messages are lost.
(The post-batch repair seek in `#batchProcessor` doesn't help: it only covers unresolved messages of the *current* payload, not the prefetched batches that `markStale` dropped from the cache.)
## Minimal repro
Deterministic against a fresh single-partition topic on any broker (`node repro.js` with a broker on `localhost:9092`); stalls at the pause point, e.g. `WEDGED: stuck at 66/120`:
```js
const { KafkaJS } = require('@confluentinc/kafka-javascript');
const N = 120;
const TOPIC = `repro-${Date.now()}`;
const delay = ms => new Promise(r => setTimeout(r, ms));
(async () => {
const kafka = new KafkaJS.Kafka({ kafkaJS: { clientId: 'repro', brokers: ['localhost:9092'] } });
const producer = kafka.producer({ 'linger.ms': 0 });
await producer.connect();
await producer.send({ topic: TOPIC, messages: Array.from({ length: N }, (_, i) => ({ value: `m${i}` })) });
await producer.disconnect();
const consumer = kafka.consumer({ kafkaJS: { groupId: `repro-${Date.now()}`, fromBeginning: true, autoCommit: true } });
await consumer.connect();
await consumer.subscribe({ topics: [TOPIC] });
let seen = 0, batches = 0, paused = false;
await consumer.run({
eachBatchAutoResolve: false,
eachBatch: async ({ batch, resolveOffset }) => {
batches++;
seen += batch.messages.length;
for (const m of batch.messages) resolveOffset(m.offset);
if (!paused && batches >= 4) { // pause once the cache is warm
paused = true;
consumer.pause([{ topic: batch.topic, partitions: [batch.partition] }]);
setTimeout(() => consumer.resume([{ topic: batch.topic, partitions: [batch.partition] }]), 500);
}
},
});
const start = Date.now();
while (seen < N && Date.now() - start < 30000) await delay(1000);
console.log(seen >= N ? `OK: all ${N}` : `WEDGED: stuck at ${seen}/${N}`);
process.exit(seen >= N ? 0 : 1);
})();
```
Notes from isolating this:
- The native (non-compat) API's pause/resume works correctly under the same sequence — the defect is in the compat layer's cache bookkeeping.
- Internal tracing shows `consume()` correctly returning 0 after resume: librdkafka has already delivered the messages; they were dropped JS-side by `markStale` with no repair seek.
## Fix
One-line-class fix + regression test in the accompanying PR: update `#lastConsumedOffsets` in `#eachBatchPayload_resolveOffsets`, mirroring the eachMessage path. With it, the repro and the new regression test pass, and the full `pause.spec.js` suite stays green (16/16).
Contributor guide
Research direction
The compat consumer logic is under lib/; start with #eachBatchPayload_resolveOffsets and compare it with #messageProcessor and #pauseInternal. Add regression coverage in pause.spec.js for a warm prefetched cache and pause/resume, then run the full pause.spec.js suite and confirm all prefetched messages are consumed.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- javascript, kafka, typescript
- Domain
- backend, distributed-systems, testing-qa
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 76/100