JITSU-214 fix(bulker): batch consumers self-heal after losing Kafka group membership

[absorbb]

Fixes JITSU-214: batch consumers that lost their Kafka consumer-group membership kept running every period, polled nothing, and looked like idle topics while events piled up (262k waiting on one topic when found). The only trace was Failed to query commited offsets. once the broker garbage-collected the empty group a week later.

How a consumer became a zombie

• Nothing detected a lost membership. A consumer that is not a group member has no assignment; ConsumeAll saw a high watermark, polled for 5 s, got nothing, and returned silently.
• Dead consumers were kept. A fatal or non-retriable ReadMessage error inside a batch only ended the batch. librdkafka marks the instance inoperable after a fatal error, so every later poll just timed out.
• Restarts fenced themselves. restartConsumer closed the old consumer asynchronously and created the new one after a fixed delay, with the same static group.instance.id. When the close was still in flight, the broker fenced one of them (FENCED_INSTANCE_ID, fatal) — 6 occurrences in a week of logs.
• No margin on the heartbeat. The paused-consumer heartbeat ran at exactly ½ of max.poll.interval.ms; one poll waiting its full timeout plus scheduling delay overshot and librdkafka left the group — 96 occurrences in a week.

Changes (bulkerapp/app/abstract_batch_consumer.go, batch_consumer.go, retry_consumer.go)

• Self-heal. In ConsumeAll, a consumer that was not created in this run, is older than a grace period (heartbeat interval + session timeout, ~145 s with defaults), has no partition assignment, and sees lag on the topic is treated as having lost membership: membership_lost metric + error log, then a synchronous restart and the run continues on the new consumer. Valid because batch topics are sharded per bulker instance, so a batch consumer is its group's only member for its partition. Retry mode is deliberately excluded: those consumers share a group across the fleet, so a missing assignment is a routine rebalance and restarting would amplify it. A zombied retry consumer still recovers through the read-error path below.
• Fence-safe restart. restartConsumer creates the new consumer first, under a fresh group.instance.id suffix (-rN), swaps it in, and closes the old one after a quarantine that outlasts any in-flight poll. Two static members with different instance ids cannot fence each other; the broker hands the partition over when the old member's session times out. Every close goes through an idempotent closeConsumer (confluent-kafka-go panics on a double Close), the suspend path only nils the pointer it looked at, restarts are serialized, and a consumer younger than a session timeout is not restarted again. Init failures retry as before.
• Fatal errors restart. A fatal or non-retriable ReadMessage error in the batch or retry read loop triggers an async restart (onReadError), and errors that mean "left the group" (MAX_POLL_EXCEEDED, FENCED_INSTANCE_ID, UNKNOWN_MEMBER_ID, any fatal) are logged as membership loss with the metric, in the read loops and in the paused heartbeat.
• Heartbeat at ⅓ of max.poll.interval.ms (paused heartbeat and the in-batch pause timer during slow loads).
• Log levels. A missing committed offset is info (a brand-new topic is the common case); a real Committed() failure stays error with a query_committed_failed metric.
• Nil-consumer guards in pauseKafkaConsumer and resume, which could dereference nil during a restart.

Metrics are on the existing ConsumerErrors counter as new errorType labels: membership_lost, query_committed_failed.

Testing

go build ./..., go vet, and unit tests for the pure helpers (hasLag, membershipLossReason). No integration test: reproducing group deletion or fencing needs a broker with a tiny offsets retention and timing control. Verification is in production: after deploy, membership_lost should fire once per zombie and the topics listed in the issue should drain; the remediation restart of bulker3-7/11/13 is no longer needed.

🤖 Generated with Claude Code

https://claude.ai/code/session_01ARxaC6w4JJ8Xw4b8Pr6b5Q

[jitsu-code-review]

Reviewed the batch-consumer membership, pause/resume, and restart paths.

Finding:

• The restart cooldown also suppresses recovery for consumers created at startup or after suspend.

[jitsu-code-review]

Reviewed the Kafka consumer recovery, pause/heartbeat, and commit paths.

Finding: restart publication can race consumer retirement and leave a subscribed replacement alive after the consumer has been closed.

[jitsu-code-review]

Reviewed the Kafka batch/retry consumer membership-recovery changes, including restart serialization, paused-consumer handling, offset commit recovery, and retirement races. I found no additional actionable issues. Existing resolved review threads were checked and not re-raised.