From edff27a8368c38eb8b6ccb48357e6200a1814165 Mon Sep 17 00:00:00 2001 From: Andrew Houghton Date: Wed, 10 Jun 2026 11:45:36 -0700 Subject: [PATCH 1/2] fix: fail over from a hung bootstrap broker during InitProducerId #initIdempotentProducer and #findCoordinator never passed `attempt` to kGetBootstrapConnection, so its retry-rotation branch stayed dead and an idempotent producer kept retrying InitProducerId against a reachable-but-hung bootstrap broker instead of failing over to known-healthy brokers. Thread `attempt` through both paths. Fixes #317. Signed-off-by: Andrew Houghton --- src/clients/producer/producer.ts | 23 ++++++---- test/clients/producer/producer.test.ts | 62 ++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 9 deletions(-) diff --git a/src/clients/producer/producer.ts b/src/clients/producer/producer.ts index 18b5ddee..5ebad2c5 100644 --- a/src/clients/producer/producer.ts +++ b/src/clients/producer/producer.ts @@ -601,7 +601,7 @@ export class Producer { this[kPerformWithRetry]( 'findCoordinator', - retryCallback => { + (retryCallback, attempt) => { this[kGetBootstrapConnection]((error, connection) => { if (error) { retryCallback(error) @@ -616,7 +616,7 @@ export class Producer { if (error) { @@ -909,12 +909,8 @@ export class Producer { this[kPerformWithRetry]( 'initProducerId', - retryCallback => { - const connector = transactionalId - ? this.#getCoordinatorConnection.bind(this) - : this[kGetBootstrapConnection].bind(this) - - connector((error, connection) => { + (retryCallback, attempt) => { + const onConnection: CallbackWithPromise = (error, connection) => { if (error) { retryCallback(error) return @@ -937,7 +933,16 @@ export class Producer kGetBootstrapConnection). The idempotent path must thread + // `attempt` so kGetBootstrapConnection rotates past a reachable-but-hung broker. + if (transactionalId) { + this.#getCoordinatorConnection(onConnection) + } else { + this[kGetBootstrapConnection](onConnection, attempt) + } }, (error, response) => { if (error) { diff --git a/test/clients/producer/producer.test.ts b/test/clients/producer/producer.test.ts index 6441b67d..b4038f21 100644 --- a/test/clients/producer/producer.test.ts +++ b/test/clients/producer/producer.test.ts @@ -14,6 +14,8 @@ import { import type { ClusterMetadata } from '../../../src/clients/base/types.ts' import { baseMetadataChannel, + type Broker, + type CallbackWithPromise, type ClientDiagnosticEvent, compressionsAlgorithms, ConfluentSchemaRegistry, @@ -711,6 +713,66 @@ test('initIdempotentProducer should handle unavailable API errors', async t => { ) }) +test('initIdempotentProducer fails over to a healthy broker when the bootstrap broker is reachable but hung', async t => { + // Regression: an idempotent producer running InitProducerId must rotate past a + // bootstrap broker that accepts TCP but never serves requests (paused/fenced/GC-stalled). + // The rotation in kGetBootstrapConnection only fires when attempt > 0 is threaded + // through, which #initIdempotentProducer previously failed to do, so it wedged on the + // hung broker for the whole retry budget instead of failing over. + const producer = createProducer(t, { + idempotent: true, + bootstrapBrokers: ['broker-a:9092', 'broker-b:9093', 'broker-c:9094'], + retries: 5, + retryDelay: 0 + }) + + // Every broker accepts TCP, so getFirstAvailable always returns a connection to whatever + // broker sits at the head of the list. The "hung" broker only manifests once InitProducerId + // is actually executed against it. + const hungHost = 'broker-a' + const triedHeads: string[] = [] + const pool = producer[kConnections] + pool.getFirstAvailable = function (brokers: Broker[], callback: CallbackWithPromise) { + const head = brokers[0] + triedHeads.push(head.host) + callback(null, { host: head.host } as Connection) + } as typeof pool.getFirstAvailable + + mockMethod(producer, kGetApi, () => true, null, undefined, (original, name, callback) => { + if (name === 'InitProducerId') { + const api = ( + connection: Connection, + _transactionalId: string | undefined, + _timeout: number, + _producerId: bigint, + _producerEpoch: number, + apiCallback: Callback + ) => { + if (connection.host === hungHost) { + // TCP-reachable but never responds: surfaces as a retriable connection reset. + apiCallback(new NetworkError('Connection closed')) + return + } + + apiCallback(null, { throttleTimeMs: 0, errorCode: 0, producerId: 42n, producerEpoch: 0 }) + } + + callback(null, api as any) + return true + } + + original(name, callback) + return true + }) + + const info = await producer.initIdempotentProducer({}) + + strictEqual(info.producerId, 42n) + // attempt 0 selected the hung bootstrap broker; attempt 1 rotated past it to a healthy one. + strictEqual(triedHeads[0], 'broker-a') + strictEqual(triedHeads[1], 'broker-b') +}) + test('send should return ProduceResult with offsets and support diagnostic channels', async t => { const producer = createProducer(t) const testTopic = await createTopic(t) From 5ee3904dc85700150d72e6687d2ae6aa2521308d Mon Sep 17 00:00:00 2001 From: Andrew Houghton Date: Mon, 15 Jun 2026 14:34:47 -0700 Subject: [PATCH 2/2] test: use mockConnectionPoolGetFirstAvailable in failover test Replace hand-rolled getFirstAvailable override with the existing mock helper. Addresses review feedback on #318. Signed-off-by: Andrew Houghton --- test/clients/producer/producer.test.ts | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/test/clients/producer/producer.test.ts b/test/clients/producer/producer.test.ts index b4038f21..0827a9ca 100644 --- a/test/clients/producer/producer.test.ts +++ b/test/clients/producer/producer.test.ts @@ -14,8 +14,6 @@ import { import type { ClusterMetadata } from '../../../src/clients/base/types.ts' import { baseMetadataChannel, - type Broker, - type CallbackWithPromise, type ClientDiagnosticEvent, compressionsAlgorithms, ConfluentSchemaRegistry, @@ -731,12 +729,12 @@ test('initIdempotentProducer fails over to a healthy broker when the bootstrap b // is actually executed against it. const hungHost = 'broker-a' const triedHeads: string[] = [] - const pool = producer[kConnections] - pool.getFirstAvailable = function (brokers: Broker[], callback: CallbackWithPromise) { + mockConnectionPoolGetFirstAvailable(producer[kConnections], () => true, null, undefined, (_original, brokers, callback) => { const head = brokers[0] triedHeads.push(head.host) callback(null, { host: head.host } as Connection) - } as typeof pool.getFirstAvailable + return true + }) mockMethod(producer, kGetApi, () => true, null, undefined, (original, name, callback) => { if (name === 'InitProducerId') {