confluentinc / confluentinc/confluent-kafka-javascript

pause() during graceful shutdown corrupts committed offsets causing message reprocessing

Open
#404 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
TypeScript
Stars
304
Forks
45
Avg merge
11h 47m
Merged PRs (30d)
5

Description

**Description**

Calling `pause()` during graceful shutdown causes offset corruption when `enable.auto.commit: true`. The consumer seeks to stale cached offsets and commits them, overwriting correct auto-committed offsets. This results in already-processed messages being reprocessed after restart or rebalance.

**Environment Information**

- OS: macOS (reproducible on Linux)
- Node Version: 23.x
- NPM Version: 9.x / 10.x
- C++ Toolchain: clang/g++
- confluent-kafka-javascript version: 1.6.0

**Steps to Reproduce**

1. Create consumer group with multiple workers (`enable.auto.commit: true`)
2. Consume messages actively (e.g., 1000 msg/sec)
3. During consumption, trigger a rebalances and initiate graceful shutdown that calls `pause()`:
```javascript
await consumer.pause([{ topic: 'my-topic' }]);
await waitForInflightMessages(); // Wait for handlers to complete
await consumer.disconnect();
```
4. Trigger sequential shutdowns (e.g., rolling deployment) causing rebalances
5. Remaining/restarted consumers reprocess already-consumed messages

**Root Cause**

In `lib/kafkajs/_consumer.js`:

1. **`#pauseInternal()` (line 1838-1862)**: Reads from `#lastConsumedOffsets` cache
2. **Stale cache**: `#lastConsumedOffsets` is never cleaned during rebalances (line 359-369), contains outdated offsets
3. **`#seekInternal()` commits** (line 1821-1822): Seeks to stale offset and commits it immediately when `enable.auto.commit: true`
4. **Race with auto-commit**: Overwrites correct offsets stored via `_offsetsStoreSingle()` but not yet auto-committed

```javascript
// Line 1848-1853: Uses stale cached offset
if (this.#lastConsumedOffsets.has(key)) {
const seekOffset = this.#lastConsumedOffsets.get(key);
// ...seeks to seekOffset.offset + 1
}

// Line 1821-1822: Commits the stale offset
if (offsetsToCommit.length !== 0 && this.#internalConfig['enable.auto.commit']) {
await this.#commitOffsetsUntilNoStateErr(offsetsToCommit); // ← COMMITS STALE OFFSET
}
```

**Expected Behavior**

Graceful shutdown should not regress committed offsets. Auto-commit should handle offset management based on `_offsetsStoreSingle()` calls.

**Actual Behavior**

`pause()` immediately commits stale cached offsets, overwriting correct offsets, causing message reprocessing after restart.

**confluent-kafka-javascript Configuration Settings**

```javascript
{
'group.id': 'my-consumer-group',
'enable.auto.commit': true,
'auto.commit.interval.ms': 5000,
// ... other settings
}
```

**Additional context**

- Issue occurs most frequently during rolling deployments with sequential shutdowns
- Affects random partitions unpredictably (depends on which cached offsets are stale)
- `#lastConsumedOffsets` Map has no lifecycle management - entries never expire or get cleaned during rebalances
- The `pause()` method's seek-and-commit behavior conflicts with graceful shutdown scenarios

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.