confluentinc / confluentinc/confluent-kafka-javascript

Cross-topic blocking still present in v1.8.2 (related to #247)

Open
#468 1 comment 1 reaction 0 assignees View on GitHub
Dominant language
TypeScript
Stars
304
Forks
45
Avg merge
11h 47m
Merged PRs (30d)
5

Description

## Description

Cross-topic blocking reported in #247 is still present in v1.8.2. When a single consumer group subscribes to multiple topics, slow message processing on one topic completely blocks consumption from other topics.

## Benchmark Results (v1.8.2, Kafka 3)

| Metric | Value |
|--------|-------|
| topic-fast completion (from publish) | **95.32s** |
| topic-slow processed (at fast complete) | 1000/1000 |
| topic-slow progress | 100.0% |
| Blocked? | **YES - cross-topic blocking detected** |
| partitionsConsumedConcurrently | 2 |

**Environment**: confluent-kafka-javascript v1.8.2, Kafka 3, Node.js

## Reproduction Scenario

1. Subscribe to 2 topics with a single consumer group: `topic-slow` (1 partition), `topic-fast` (1 partition)
2. Publish 1000 messages to `topic-slow`, start consuming with 100ms/msg handler delay
3. After 50 slow messages processed, publish 10 messages to `topic-fast` with instant handler
4. Measure time for `topic-fast` to complete all 10 messages

**Expected**: `topic-fast` completes in <1s (independent consumption with `partitionsConsumedConcurrently: 2`)
**Actual**: `topic-fast` took **95.32s**, waiting for all 1000 slow messages to finish first

## Root Cause Analysis

The blocking originates in `lib/kafkajs/_consumer.js`, specifically in `#consumeSingleCached` and `#consumeCachedN`:

```javascript
// lib/kafkajs/_consumer.js — #consumeSingleCached
async #consumeSingleCached(ppc) {
const msg = this.#messageCache.next(ppc);
if (msg) {
return msg;
}

// THIS CONDITION CAUSES CROSS-TOPIC BLOCKING:
if (!msg && this.#messageCache.assignedSize !== 0) {
await this.#messageCache.availablePartitions();
return null;
}

return this.#fetchAndResolveWith(() => this.#messageCache.next(),
this.#messageCacheMaxSize);
}
```

**Why it blocks**: When `partitionsConsumedConcurrently >= 2`, each worker gets a PerPartitionCache (PPC). If WorkerA is processing `topic-slow` (holding its PPC), WorkerB finishes `topic-fast` quickly but then:
1. `this.#messageCache.next(ppc)` returns `null` (no more cached messages)
2. `this.#messageCache.assignedSize !== 0` is `true` (WorkerA still holds slow PPC)
3. WorkerB awaits `availablePartitions()` — **blocked until WorkerA returns its PPC**
4. New `topic-fast` messages cannot be fetched until `topic-slow` processing completes

The same pattern exists in `#consumeCachedN`.

## Note on Fix Complexity

Simply removing the `assignedSize !== 0` condition causes rebalance test failures (`consumerCacheTests.spec.js` — "is cleared before rebalance"). Additional fetches cause one consumer to consume all messages from librdkafka before rebalance, leaving none for the joining consumer.

A proper fix likely requires architectural changes such as:
- Per-topic or per-partition fetch scheduling
- Allowing idle workers to fetch without blocking, while preserving rebalance safety
- Timeout-based fallback with rebalance awareness

## Related

- #247

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.