Consumer silently stops consuming — race conditions in #nonEmpty and #fetchInProgress signals

[ndrantotiana]

Description

The KafkaJS compat consumer silently stops calling eachBatch/eachMessage after a period of inactivity. The consumer stays connected (heartbeats continue, partitions assigned), but the internal worker loop never fetches messages again. Only a full disconnect() + new consumer instance recovers.

Reproduction

1. Create a consumer with eachBatch handler and autoCommit: false
2. Produce a burst of messages, let the consumer process them
3. Wait for max.poll.interval.ms (default 300s) with no new messages
4. Produce a new message during the cache expiration window
5. The consumer never calls eachBatch for the new message

This is intermittent — it depends on the exact timing of the message arrival relative to the cache expiration cycle.

Environment

• @confluentinc/kafka-javascript: 1.8.2
• Node.js: 18
• OS: Linux (Kubernetes pods)
• Kafka: Amazon MSK

Root cause analysis

The issue is a race condition between #cacheExpirationLoop and #queueNonEmptyCb in lib/kafkajs/_consumer.js.

Normal flow (working)

1. Workers are running, waiting on await this.#nonEmpty in #fetchAndResolveWith (line 1054)
2. librdkafka C thread receives a message, calls #queueNonEmptyCb
3. #queueNonEmptyCb → #notifyNonEmpty() → resolves #nonEmpty, sets it to null
4. Worker wakes up, calls consume(), gets the message, processes it

Stall flow (broken)

1. No messages for max.poll.interval.ms (= #cacheExpirationTimeoutMs, line 709)
2. #cacheExpirationLoop detects timeout (line 1581: now > cacheExpirationTimeout)
3. Calls #addPendingOperation(() => #clearCacheAndResetPositions()) (line 1582)
4. #addPendingOperation → #resolveWorkerTerminationScheduled() (line 1781) — terminates all workers
5. #resolveWorkerTerminationScheduled also calls #queueNonEmptyCb() (line 1624) — resolves #nonEmpty, sets it to null
6. Workers exit their while loop (line 1517: !this.#workerTerminationScheduled.resolved)
7. Meanwhile: a new message arrives, librdkafka calls #queueNonEmptyCb → #notifyNonEmpty() → #nonEmpty is already null (from step 5) → no-op, signal lost
8. #runInternal continues: await Promise.allSettled(workers) → await cacheExpirationLoop → #executePendingOperations → #clearCacheAndResetPositions (seeks back to last consumed offset)
9. New workers spawned
10. Worker calls #consumeCachedN → cache empty → #fetchAndResolveWith
11. #nonEmpty is null → proceeds to consume()
12. consume() returns the message (seek reset position, librdkafka re-fetches) — this CAN work
13. But if consume() returns 0 (race with librdkafka's internal re-fetch): #messageCache.assignedSize === 0 → #nonEmpty = new DeferredPromise() (line 1096)
14. Worker waits on #nonEmpty forever — the #queueNonEmptyCb signal was already spent in step 7
15. Consumer appears alive (heartbeats via librdkafka C thread) but never processes messages again

Key observation

#nonEmpty is a one-shot signal. Once resolved, it's set to null. If a new message arrives while #nonEmpty is null (between steps 5 and 13), the signal is lost. The worker creates a new DeferredPromise in step 13 but #queueNonEmptyCb has already fired — it won't fire again until ANOTHER new message arrives.

Evidence

We observed this across 19 microservices using the KafkaJS compat layer:

• Consumer processes messages successfully for 20+ minutes
• eachBatch stops being called (confirmed via periodic heartbeat logging: callsSinceLast=0, growing silenceMs)
• Consumer stays connected (librdkafka heartbeats, partitions assigned)
• No errors logged by either our code or the consumer
• Kafka lag grows (messages produced but never consumed)
• Only creating a brand new consumer instance recovers consumption

The stall consistently occurs after a quiet period matching max.poll.interval.ms.

Proposed fix

Add a timeout to await this.#nonEmpty in #fetchAndResolveWith (line 1054):

// BEFORE (line 1053-1058):
if (this.#nonEmpty) {
    await this.#nonEmpty;
    return null;
}

// AFTER:
if (this.#nonEmpty) {
    await Timer.withTimeout(1000, this.#nonEmpty);  // 1s fallback, arbitrary
    return null;
}

This ensures that even if the #nonEmpty signal is lost during the cache expiration race, the worker retries the fetch after 10 seconds instead of waiting forever. Timer.withTimeout is already used throughout the codebase (lines 1553, 1600).

The 10-second delay is negligible compared to the current "stall forever" behavior, and consume() will return 0 if there are truly no messages (the worker loops back and waits again).

Workaround

We implemented a stall detector that creates a brand new consumer instance (disconnect old → create new → connect → subscribe → run) after 10 minutes of silence. This works because disconnect() sets #disconnectStarted which stops the #runInternal loop, and a new consumer has fresh state.

Note: reconnecting the SAME consumer after disconnect() fails because #running is set to true in run() (line 1305) but never reset to false in disconnect() (line 2132 sets #disconnectStarted but not #running=false). This is a separate issue.

[ndrantotiana]

Update: Second stall vector found — #fetchInProgress

After deploying the #nonEmpty timeout patch, we still observed stalls in production (10+ minutes of silence despite the 1s timeout). Investigation revealed a second stall vector in the same #fetchAndResolveWith method.

Root cause

#fetchAndResolveWith has two blocking awaits before the actual consume() call:

async #fetchAndResolveWith(takeFromCache, size) {
    if (this.#fetchInProgress) {
      await this.#fetchInProgress;   // ← STALL VECTOR 2 (no timeout)
      return null;
    }

    if (this.#nonEmpty) {
      await this.#nonEmpty;          // ← STALL VECTOR 1 (patched with 1s timeout)
      return null;
    }
    // ...
    this.#fetchInProgress = new DeferredPromise();
    this.#internalClient.consume(size, (err, messages) => fetchResult.resolve([err, messages]));
    [err, messages] = await fetchResult;   // ← STALL VECTOR 3 (C-level, not patchable from JS)
    // ...
    finally {
      this.#fetchInProgress.resolve();
    }
}

If this.#internalClient.consume() hangs at the C level (librdkafka), #fetchInProgress is never resolved. Other workers waiting on await this.#fetchInProgress (line 1047) are blocked forever — our #nonEmpty patch doesn't help because the worker never reaches that check.

Evidence

With only the #nonEmpty patch deployed, we observed:

• Last eachBatch END OK at 14:31:24
• Consumer heartbeat shows callsSinceLast=0 from 14:32:05 onwards
• Stall detector fires at 14:42:05 (10.8 min silence)
• New consumer instance created → resumes consuming immediately

The #nonEmpty timeout (1s) was active but the worker was stuck at await this.#fetchInProgress instead.

Updated fix

We extended the patch to cover both await points:

// Patch 1 (original): #nonEmpty timeout
if (this.#nonEmpty) {
    await Timer.withTimeout(1000, this.#nonEmpty);
    return null;
}

// Patch 2 (new): #fetchInProgress timeout
if (this.#fetchInProgress) {
    await Timer.withTimeout(1000, this.#fetchInProgress);
    return null;
}

Both use Timer.withTimeout(1000, ...) — consistent with the codebase's existing pattern (lines 1553, 1600). Workers retry every 1s instead of waiting forever.

Additional note: #running flag not reset on disconnect

As mentioned in the original report, disconnect() (line 2132) sets #disconnectStarted but never resets #running to false. This prevents reconnecting the same consumer instance after a stall — run() throws "Consumer is already running" (line 1302). Our workaround creates a brand new consumer instance, but fixing #running in disconnect() would be the proper solution.

[ndrantotiana]

Third stall vector: cache expiration during eachBatch processing

#lastFetchClockNs is set once in #runInternal (line 1652) when workers are spawned, but never updated after eachBatch completes. If eachBatch processing takes longer than cacheExpirationTimeoutMs, the #cacheExpirationLoop (line 1581) fires #clearCacheAndResetPositions() which seeks all partitions back to lastConsumedOffsets.

This causes offset loss: offsets resolved by eachBatch (via resolveOffset + commitOffsetsIfNecessary) may not have persisted to the broker before the seek resets positions. When the stall detector creates a new consumer instance, it starts from the broker's committed offset (behind the reset position) → re-delivers already-processed messages.

Repro: consumer with eachBatch handler that takes >1s (tokenization, S3 upload). After eachBatch returns, the worker calls fetcher() again. The #cacheExpirationLoop detects cache expiration → #clearCacheAndResetPositions → seeks back → consumer stalls waiting for the next fetch cycle → stall detector fires → new consumer → re-delivery.

Suggested fix: update #lastFetchClockNs = hrtime.bigint() after await config.eachBatch(payload) returns in #batchProcessor, so the cache expiration timer restarts from the eachBatch completion time.

[ndrantotiana]

Fourth stall vector: #fetchInProgress not timed out in #queueNonEmptyCb

#queueNonEmptyCb() (line 1489-1499) awaits this.#fetchInProgress without timeout:

#queueNonEmptyCb() {
    const nonEmptyAction = async () => {
      if (this.#fetchInProgress)
        await this.#fetchInProgress;  // ← no timeout
      this.#notifyNonEmpty();
    };

If #fetchInProgress never resolves (same root cause as vector 2 — consume() hangs in librdkafka), #notifyNonEmpty() is never called → #nonEmpty is never signaled → workers waiting on #nonEmpty in #fetchAndResolveWith stall forever.

This is a separate code path from #fetchAndResolveWith (vector 2). #queueNonEmptyCb is called from the cache expiration cycle when the cache needs to notify workers that new messages may be available. Even with vector 2 patched, this path can independently stall the consumer.

Suggested fix: add timeout to the await, consistent with #fetchAndResolveWith:

if (this.#fetchInProgress)
    await Timer.withTimeout(1000, this.#fetchInProgress);

---

Updated third vector (cache expiration during eachBatch):

#lastFetchClockNs is set once in #runInternal (line 1652) when the worker loop starts, but never updated after eachBatch completes in #batchProcessor. If eachBatch processing exceeds cacheExpirationTimeoutMs, the #cacheExpirationLoop (line 1581) triggers #clearCacheAndResetPositions() which seeks all partitions back to #lastConsumedOffsets.

This causes committed offset loss: after eachBatch resolves offsets and calls commitOffsetsIfNecessary, the cache expiration loop fires, seeks positions backward, and the commit may not have persisted yet. On stall recovery (new consumer instance), the broker's committed offset is behind → re-delivers already-processed messages.

Suggested fix: update #lastFetchClockNs = hrtime.bigint() after await config.eachBatch(payload) returns in #batchProcessor, so the cache expiration timer restarts from the processing completion time rather than the fetch time.

---

Summary of all vectors reported:

1. #nonEmpty signal lost during #cacheExpirationLoop → workers wait forever on await this.#nonEmpty
2. #fetchInProgress hangs in #fetchAndResolveWith → workers wait forever on await this.#fetchInProgress
3. Cache expires during long eachBatch → #clearCacheAndResetPositions seeks back → offsets lost
4. #fetchInProgress hangs in #queueNonEmptyCb → #notifyNonEmpty never called → workers stall on #nonEmpty

All four are independently reproducible with consumers that have eachBatch handlers taking >1s (common with gRPC/S3 operations). Suggested fixes: 1s timeouts on vectors 1, 2, 4 and #lastFetchClockNs reset for vector 3.

[ndrantotiana]

confluent-stall-repro.js

Minimal reproduction script + two additional stall vectors

Attaching a minimal reproduction script that demonstrates the stall. Run with:

# Default (5s processing per batch)
node confluent-stall-repro.js

# Trigger vector 3 faster (30s processing)
PROCESSING_TIME_MS=30000 node confluent-stall-repro.js

The script creates a consumer with eachBatch + autoCommit: false + manual resolveOffset/commitOffsetsIfNecessary. It simulates real-world processing time (gRPC calls, S3 uploads). After the first batch completes, the consumer silently stalls — no errors, no eachBatch calls, consumer stays connected.

Two additional stall vectors found since the original report

Vector 3: Cache expiration during eachBatch processing

#lastFetchClockNs is set once in #runInternal (line 1652) when workers spawn, but never updated after eachBatch completes in #batchProcessor. If eachBatch takes longer than cacheExpirationTimeoutMs, the #cacheExpirationLoop (line 1581) fires #clearCacheAndResetPositions() which seeks all partitions back to #lastConsumedOffsets.

This causes offset loss: offsets resolved by eachBatch via resolveOffset + commitOffsetsIfNecessary may not have persisted to the broker before the seek resets positions. When a new consumer instance is created after stall recovery, it starts from the broker's committed offset (behind) and re-delivers already-processed messages.

Suggested fix — update the clock after eachBatch returns:

// In #batchProcessor, after await config.eachBatch(payload):
this.#lastFetchClockNs = hrtime.bigint();

Vector 4: #fetchInProgress not timed out in #queueNonEmptyCb

#queueNonEmptyCb() (line 1489) awaits this.#fetchInProgress without timeout:

#queueNonEmptyCb() {
    const nonEmptyAction = async () => {
      if (this.#fetchInProgress)
        await this.#fetchInProgress;  // ← no timeout, can hang forever
      this.#notifyNonEmpty();
    };

If #fetchInProgress never resolves (same root cause as vector 2 — consume() hangs in librdkafka), #notifyNonEmpty() is never called → workers waiting on #nonEmpty stall forever. This is a separate code path from #fetchAndResolveWith (vector 2), so patching vector 2 alone doesn't cover this.

Suggested fix:

if (this.#fetchInProgress)
    await Timer.withTimeout(1000, this.#fetchInProgress);

Summary of all 4 vectors

#	Location	Root cause	Suggested fix
1	#fetchAndResolveWith	#nonEmpty signal lost during cache expiration	Timer.withTimeout(1000, this.#nonEmpty)
2	#fetchAndResolveWith	#fetchInProgress hangs (consume() never returns)	Timer.withTimeout(1000, this.#fetchInProgress)
3	#batchProcessor	#lastFetchClockNs not updated after eachBatch → cache expires → positions reset	this.#lastFetchClockNs = hrtime.bigint() after eachBatch
4	#queueNonEmptyCb	#fetchInProgress hangs → #notifyNonEmpty never called	Timer.withTimeout(1000, this.#fetchInProgress)

Observed across 19 microservices in production. All 4 vectors are independently reproducible with consumers that have eachBatch handlers taking >1s.

[ndrantotiana]

Vector 5: availablePartitions() infinite loop after cooperative-sticky rebalance

We found an additional stall vector not covered by the original patches. This one is triggered specifically by rebalances (rolling deployments, scaling up/down, cooperative-sticky partition reassignment).

Root cause

In #consumeSingleCached (line ~1122) and #consumeCachedN (line ~1152):

if (!msgs && this.#messageCache.assignedSize !== 0) {
  await this.#messageCache.availablePartitions();
  return null;
}

return this.#fetchAndResolveWith(/* ... */);

After a cooperative-sticky rebalance assigns partitions to a new consumer instance (or after worker respawn), assignedSize > 0 but the message cache is empty (size === 0). No messages have been fetched yet for the newly assigned partitions.

The worker enters an infinite loop:

1. nextN() → null (cache empty)
2. assignedSize !== 0 → true (partitions assigned via rebalance)
3. await availablePartitions() → resolves after ~1s (from #cacheExpirationLoop calling #notifyNonEmpty)
4. Returns null → loops back to step 1

consume() is never called because the worker never reaches #fetchAndResolveWith. Messages are never fetched. eachBatch is never invoked. The consumer appears alive (heartbeats continue) but permanently stalled.

Why patches 1-4 don't help

Patches 1-4 add timeouts to #nonEmpty and #fetchInProgress inside #fetchAndResolveWith. But the worker never enters #fetchAndResolveWith — the assignedSize !== 0 check short-circuits to availablePartitions().

Suggested fix

The availablePartitions() wait is intended for when other workers are busy processing cached messages on assigned partitions. When the cache is empty (size === 0), the worker should fall through to #fetchAndResolveWith to populate the cache:

// Before (stalls when cache empty after rebalance):
if (!msgs && this.#messageCache.assignedSize !== 0) {

// After (falls through to fetch when cache is empty):
if (!msgs && this.#messageCache.assignedSize !== 0 && this.#messageCache.size > 0) {

Same fix applies to both #consumeSingleCached and #consumeCachedN.

Production evidence

Observed across all pods simultaneously during a rolling deployment of a normalizer service (6 pods):

13:04:50  New pods start, consumer.run() called
13:06:54  Old pods terminate → cooperative-sticky rebalance assigns partitions to new pods
13:10:22  Brief activity (a few eachBatch calls), then silence
13:12:53  Pod 1: stall detected — silenceMs=226,502ms, stallRecoveryCount=1
13:13:33  Pod 2: stall detected — silenceMs=194,655ms
13:13:40  Pod 3: stall detected — silenceMs=198,519ms
13:13:40  Pod 4: stall detected — silenceMs=199,651ms
13:14:05  Pod 5: stall detected — silenceMs=226,891ms

All 6 pods stalled after the rebalance. Our stall detector (external watchdog) created new consumer instances to recover. Consumer version: @confluentinc/kafka-javascript with all 4 patches from the original report applied and verified in the running pod's _consumer.js.

Here is the updated reproduction script with 2 scenarios:

node scripts/confluent-stall-repro.js              # Vectors 1-4 (cache expiration)
node scripts/confluent-stall-repro.js --rebalance   # Vector 5 (rebalance stall)

confluent-stall-repro.js