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

[beygee]

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:

// 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

• Consumption rate of messages from one topic effects the rate of consumption from another topic #247

[beygee]

Reproduction Benchmark Code

Here's the core logic of the benchmark used to reproduce this issue:

const SLOW_MSG_COUNT = 1000;
const FAST_MSG_COUNT = 10;
const SLOW_HANDLER_DELAY_MS = 100;

// 1. Setup: Create 2 topics (1 partition each), publish 1000 slow messages first
// 2. Start consumer subscribing to both topics
const consumer = kafka.consumer({
  kafkaJS: { groupId, fromBeginning: true },
});
await consumer.subscribe({ topics: ['topic-fast', 'topic-slow'] });

let fastConsumed = 0, slowConsumed = 0;
let fastPublishTs = 0, fastPublished = false;

consumer.run({
  partitionsConsumedConcurrently: 2,  // enable concurrent partition processing
  eachMessage: async ({ topic }) => {
    if (topic === 'topic-slow') {
      await sleep(100);  // simulate slow processing (100ms per message)
      slowConsumed++;

      // After 50 slow messages processed (~5s in), publish 10 fast messages
      // This ensures fast messages arrive WHILE slow processing is ongoing
      if (slowConsumed === 50 && !fastPublished) {
        fastPublished = true;
        fastPublishTs = performance.now();
        publishFastMessages();  // publish 10 messages to topic-fast
      }
    } else if (topic === 'topic-fast') {
      fastConsumed++;  // instant processing, no delay

      if (fastConsumed === FAST_MSG_COUNT) {
        const elapsed = performance.now() - fastPublishTs;
        // Expected: ~1s (independent consumption)
        // Actual:   ~95s (blocked until all 1000 slow messages finish)
        console.log(`topic-fast completed in ${elapsed}ms`);
      }
    }
  },
});

Key Design Decisions

1. Slow messages published first → consumer starts processing topic-slow before topic-fast messages exist
2. Fast messages published mid-stream (after 50 slow messages) → guarantees the blocking condition is active when fast messages arrive
3. partitionsConsumedConcurrently: 2 → should allow WorkerA on topic-slow and WorkerB on topic-fast simultaneously
4. Threshold: topic-fast taking >10s = blocking detected (expected <1s)

Result

With v1.8.2, topic-fast took 95.32s — it had to wait for all 1000 topic-slow messages to finish before it could even start fetching, despite partitionsConsumedConcurrently: 2 being set.