eachBatch: cache expiration discards already-fetched, undelivered records and skips the position rewind → silent message loss

**Environment Information**

- OS: Linux (Debian 12, x86_64)
- Node Version: 22.22.0
- NPM Version: n/a — Yarn 4.12.0
- C++ Toolchain: none (prebuilt binary from npm, no local compile)
- confluent-kafka-javascript version: **1.9.0 and 1.10.0 — both affected, identical behaviour**
- Broker: Apache Kafka 3.x, single node, mTLS
- API: kafkajs-compat (`KafkaJS.Kafka`), `eachBatch`, `autoCommit: true`, READ_COMMITTED

---

**Steps to Reproduce**

A consumer whose `eachBatch` handler takes minutes per batch permanently loses records:
they are fetched into the client's message cache, never delivered to the handler, then
discarded while the consumer offset advances past them. No error, no warning, no rebalance.

1. Create a topic with 20 partitions.
2. Start a consumer with the configuration below, whose `eachBatch` takes ~4m40s per batch
of 20 (e.g. 3 messages at a time, 40s each).
3. Produce **200 records to a single partition** in one transaction, commit, then
**disconnect the producer and never produce again**.
4. Wait.

Deterministic — five consecutive runs stopped at exactly the same record, across both
versions.

Observed:

```
delivered: 82 / 200 stranded: 118
p0: committed=82 high=201 lag=119
```

Producing one further record to that partition wakes the consumer, but the batch delivered
contains **only the new record**, and the committed offset jumps straight past everything
that was never delivered:

```
newlyDelivered: ["0:201"] # only the new record
p0: committed=202 high=202 lag=0 # offsets 82..200 committed, never delivered
```

Offsets 82–200 are lost to that consumer group.

---

**confluent-kafka-javascript Configuration Settings**

```js
const consumer = kafka.consumer({
kafkaJS: {
groupId,
fromBeginning: true,
autoCommit: true,
readUncommitted: false, // READ_COMMITTED
sessionTimeout: 6000,
heartbeatInterval: 1500,
maxBytes: 8192,
maxBytesPerPartition: 8192,
},
'max.poll.interval.ms': 900000,
'js.consumer.max.batch.size': 20,
'fetch.wait.max.ms': 5000,
});

await consumer.run({
partitionsConsumedConcurrently: 20,
eachBatch: async ({ batch }) => { /* ~4m40s per batch of 20 */ },
});
```

Producer: `acks: -1`, transactional, one transaction for all 200 records.

---

**Additional context**

*Instrumented trace.* We patched `lib/kafkajs/_consumer.js` and `_consumer_cache.js` with
tracing. Over one reproduction:

```
consume-request 674 # 3 productive, 671 returning count:0
park-wait-enter 13266
park-wait-released 13266 # balanced — no worker is blocked on #nonEmpty
valve-1s-fire 10798
cache-expiration-break 1
clearCache-skip-rewind 20
cache:nextN-null 220331 # respawned workers hot-spin on an empty cache
```

The three productive fetches:

```
consume-result {"count":1, "offsets":["0:0"]}
consume-result {"count":1, "offsets":["0:1"]}
consume-result {"count":198, "offsets":["0:2","0:3","0:4"]} # all 200 now cached
... 670 x consume-result {"count":0} # correct: broker is drained
```

So **librdkafka delivered every record into the JS message cache within 90 seconds**, and
the later empty fetches are correct. `debug=consumer,fetch,cgrp` shows no pause, no EOF, no
backoff and no offset reset. The broker is not involved.

*Timeline.*

```
14:08:19 cache:nextN-served {"served":20,"remaining":178} ┐
14:12:59 cache:nextN-served {"served":20,"remaining":158} │ steady, 4m40s per batch
14:17:39 cache:nextN-served {"served":20,"remaining":138} │
14:22:19 cache:nextN-served {"served":20,"remaining":118} ┘ last successful serve
14:23:19 cache-expiration-break <- clear queued, workers stopped
14:26:59 batch-eachBatch-returned / worker-exit <- slow batch finally returns
14:26:59 clearCacheAndResetPositions {"assignedSize":0,"lastConsumedKeys":[]}
<- 118 cached records destroyed,
rewind skipped for ALL partitions
14:37:27 cache:nextN-served {"served":1,"remaining":0} <- a newly produced record only
```

*Cause.*

1. While a batch runs for minutes no fetches happen, so `#lastFetchClockNs` goes stale.
2. `#cacheExpirationLoop` takes its `now > cacheExpirationTimeout` branch, queues
`#clearCacheAndResetPositions` via `#addPendingOperation`, and breaks — which also removes
its own 1s `#notifyNonEmpty()` valve.
3. Pending operations run only after `await Promise.allSettled(this.#workers)`, so the clear
executes the instant the slow batch returns.
4. `#clearCacheAndResetPositions` calls `this.#messageCache.clear()` — discarding every record
still cached but not yet delivered — and then, per partition:

```js
const key = partitionKey(topicPartition);
if (!this.#lastConsumedOffsets.has(key))
continue; // <-- no rewind
```

In our trace `#lastConsumedOffsets` is **empty** (`lastConsumedKeys: []`), so the rewind is
skipped for every partition, librdkafka's position stays where it had already fetched to,
the discarded records are never re-fetched, and auto-commit advances over them.

`#lastConsumedOffsets` is populated only in `#returnMessages()` — the pending-operation path
where messages are handed *back* to the cache. On this path they are dropped instead, so
nothing records where to rewind to.

*Patch we are running (not upstream-quality, but it fixes the loss).* Capture each
partition's oldest undelivered record before `clear()` and use it as the rewind point when
`#lastConsumedOffsets` has no entry:

```js
// _consumer_cache.js — PerPartitionMessageCache
_peek() { const node = this.#cache.first; return node ? node.value : null; }

// _consumer_cache.js — MessageCache
pendingHeadMessages() {
const heads = [];
for (const ppc of this.#tpToPpc.values()) {
const message = ppc._peek();
if (message) heads.push(message);
}
return heads;
}

// _consumer.js — #clearCacheAndResetPositions, before this.#messageCache.clear()
const pendingHeads = new Map();
for (const message of this.#messageCache.pendingHeadMessages()) {
const offset = Number(message.offset);
if (!Number.isFinite(offset) || offset < 0) continue;
pendingHeads.set(partitionKey(message), { offset, leaderEpoch: message.leaderEpoch });
}
// ... then per partition:
const rewindTo = this.#lastConsumedOffsets.has(key)
? this.#lastConsumedOffsets.get(key)
: pendingHeads.get(key);
if (!rewindTo) continue;
```

Result: **200/200 delivered, 0 stranded**, on the scenario that previously stopped at 82/200
in five consecutive runs. The clear becomes a re-fetch instead of a discard. Note
`#seekInternal` seeks to (and commits) exactly the offset given — an earlier version of this
patch used `offset - 1` and produced `Offset out of range` when a partition's head was at
offset 0.

*A dead end, recorded to save others the detour.* We first suspected `#nonEmpty` and bounded
the wait on it. The patched client stopped at exactly the same record; tracing then showed
`park-wait-enter == park-wait-released`, i.e. no worker was ever blocked there.

*Open question.* The expiration fired **60s** after the last fetch, although
`#cacheExpirationTimeoutMs = #maxPollIntervalMs` and we set `max.poll.interval.ms: 900000`.
Either the passthrough is not reaching `#maxPollIntervalMs`, or the timer is resolved early.
If the configured value were honoured, a 4m40s batch would not trip a 900s expiration at all,
so this may be a second, independent defect.

*Production impact.* A data-loading pipeline hung permanently with 43 of 1562 records never
delivered, requiring operator intervention. A consumer for a *different* topic in the same
process stayed healthy throughout.

A self-contained reproduction script (~250 lines, no dependencies beyond this client) is
available on request.

Read lib/kafkajs/_consumer.js and _consumer_cache.js, starting with #clearCacheAndResetPositions, pending operations, and cache expiration. Reproduce the 200-record, slow eachBatch scenario or inspect the supplied trace, then verify that expiration does not lose undelivered cached records and that positions rewind so all 200 records are delivered.

Written by the indexing model from the issue text.