KafkaJS compat: pause() on an eachBatch consumer permanently loses prefetched messages (consumption wedges)

[eqcu]

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:

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).