hcengineering/platform · warning

consumer disconnected from queue

Error message

consumer disconnected from queue

What it means

KafkaJS fires a consumer.disconnect event when the underlying consumer connection to the Kafka brokers drops. The kafka package's doConnect registers a handler that sets connected=false and logs this warning. Messages cannot be consumed until a reconnect succeeds; KafkaJS will normally auto-reconnect.

Source

Thrown at foundations/server/packages/kafka/src/index.ts:323

            await heartbeat()
            await new Promise((resolve) => setTimeout(resolve, to * retryDelay))
            if (to < maxRetryDelay) {
              to++
            }
          }
        }
      }
    })
  }

  async doConnect (): Promise<void> {
    this.cc.on('consumer.connect', () => {
      this.connected = true
      this.ctx.info('consumer connected to queue')
    })
    this.cc.on('consumer.disconnect', () => {
      this.connected = false
      this.ctx.warn('consumer disconnected from queue')
    })
    await this.cc.connect()
  }

  async doSubscribe (): Promise<void> {
    await this.cc.subscribe({
      topic: getKafkaTopicId(this.topic, this.config),
      fromBeginning: this.options?.fromBegining
    })
  }

  isConnected (): boolean {
    return this.connected
  }

  close (): Promise<void> {
    return this.cc.disconnect()
  }

View on GitHub (pinned to 63e28dc964)

Solutions

1. Check Kafka broker health and listener/advertised.listeners configuration
2. Verify network connectivity (host, port, TLS) between the server and brokers
3. Rely on KafkaJS auto-reconnect or add explicit reconnect/retry logic around consumer start
4. Review broker logs for group rebalance or auth failures that triggered the disconnect
5. Increase connectionTimeout/sessionTimeout values if timeouts cause the disconnect

Example fix

// before
await this.cc.connect()
// after
await this.cc.connect()
this.cc.on('consumer.crash', async (e) => {
  this.ctx.error('consumer crashed', { error: e })
  await this.start(this.ctx) // re-establish connection/subscription
})

const reachable = await net.connect({ host: kafkaHost, port: kafkaPort })
  .then(s => { s.destroy(); return true })
  .catch(() => false)
if (!reachable) throw new Error('Kafka broker unreachable before consumer start')

function isDisconnectEvent(e: unknown): e is { payload: { clientId: string } } {
  return typeof e === 'object' && e !== null && 'payload' in e
}

this.cc.on('consumer.disconnect', async () => {
  this.connected = false
  await backoffRetry(() => this.doConnect(), { retries: Infinity, baseMs: 1000 })
})

When it happens

Trigger: During start() -> doConnect(), the consumer's broker connection terminates — broker restart, network partition, idle connection reaped, TLS/auth failure, or group rebalancing kicking the member out.

Common situations: Kafka broker restarts or rolling upgrades; network flakiness between server and Kafka cluster; connection_max_idle_ms expiring idle connections; misconfigured advertised listeners causing unreachable broker addresses.

Related errors

• platform.status.ConnectionClosed
• Invalid endpoint reference
• No connection available
• platform.status.ConnectionClosed
• ConnectionClosed

AI-assisted analysis of hcengineering/platform@63e28dc964 (2026-08-29). Data as JSON: /api/errors/34056671d7edd41b. Report an issue: GitHub.