fix(kafka): eliminate Resume timeout race in batch consumer pause/heartbeat

[absorbb]

Summary

Two defense-in-depth fixes for the failed to resume kafka consumer: Resume timeout error seen on the retry consumer paths in bulkerapp (e.g. in.id.<workspace>.m.retry.t._all_).

• Buffer resumeChannel to size 1 so resume() (and _unpause()) can deposit the signal even when the pause heartbeat goroutine is mid-iteration. pause() drains any stale signal on entry, and the signal handler defensively re-resumes the current consumer in case it was replaced between resume()'s consumer.Resume(...) call and the heartbeat actually picking up the signal.
• Run restartConsumer asynchronously when invoked from the heartbeat via a new restartConsumerAsync helper, guarded by an atomic.Bool to prevent overlapping restart attempts. The synchronous restartConsumer entrypoint is preserved for the other call sites (pauseKafkaConsumer, etc.) that need to block until the new consumer is up.

Why this fixes the symptom

resume() and the pause-heartbeat goroutine coordinate via resumeChannel. Before this PR the channel was unbuffered, so a send only succeeds when the heartbeat is parked in its select. The heartbeat loop calls restartConsumer synchronously on non-retriable ReadMessage errors; restartConsumer blocks for KafkaSessionTimeoutMs + 15s per init attempt (default 60 s baseline) and loops if init keeps failing. With unhealthy brokers / network the heartbeat is starved past KafkaMaxPollIntervalMs (default 5 min), and resume() times out.

The same starvation also explains the sibling failed to unpause kafka consumer. from _unpause().

With the buffered channel, resume() deposits the signal immediately, calls consumer.Resume(partitions) on the current consumer, and returns. The heartbeat picks up the signal on its next pass (whenever restartConsumer finishes, or on its next ticker tick). With the async restart, the heartbeat no longer parks itself in restartConsumer at all — restart proceeds in the background, the heartbeat keeps cycling, and the new consumer is brought up paused by rebalanceCallback (which checks bc.paused) as before.

Op notes

• Behavior is unchanged on the happy path. Resume timing improves under broker degradation; the 5-minute stall in resume() is no longer reachable.
• restartConsumer itself is unchanged; only the call site in the heartbeat loop is rerouted through restartConsumerAsync. Other callers (e.g. pauseKafkaConsumer failure path) still run it synchronously.
• The new restarting atomic.Bool field is unexported and only consulted by restartConsumerAsync; piling-up overlapping restarts is suppressed (subsequent async calls are no-ops while one is in flight).

Test plan

• go vet ./bulkerapp/... passes (verified locally)
• Deploy to staging; induce a broker disconnect to force the heartbeat into the non-retriable-error branch; confirm no Resume timeout and no failed to unpause in logs after recovery
• Confirm the retry consumer (m.retry.t._all_) resumes cleanly after a forced restart
• Verify no duplicate / orphan kafka consumers when restart is triggered from heartbeat (should be at most one concurrent restartConsumer in flight per consumer)

🤖 Generated with Claude Code

[jitsu-code-review]

I found one correctness issue in this change that can regress suspend behavior under the same race this PR is addressing. Please see the inline comment.

[jitsu-code-review]

Making resumeChannel buffered removes the rendezvous semantics that _unpause() currently relies on in pauseOrSuspend() (_unpause(); consumer.Close(); bc.consumer.Store(nil)). With an unbuffered channel, _unpause() blocks until the pause heartbeat loop receives the signal; now it can return immediately while the loop is still inside ReadMessage(). Closing the consumer in that window can produce a non-timeout error in the pause loop and trigger restartConsumerAsync(), which can recreate a consumer right after we intended to suspend it. This looks like a user-visible regression (suspend may turn into restart churn).

Can we keep non-blocking resume() behavior without weakening _unpause() synchronization (e.g. separate channels/paths for resume vs suspend-ack, or an explicit ack/wait primitive for suspend)?

[absorbb]

Good catch — fixed in b45169a with the separate-channel approach you suggested.

resume() keeps the buffered resumeChannel (non-blocking deposit, which is what fixed the Resume timeout). The suspend path now uses a dedicated unbuffered stopChannel: _unpause() sends on it, so the send only completes once the heartbeat has received it and is leaving its loop — i.e. no longer inside ReadMessage(). That restores the rendezvous pauseOrSuspend relies on, so consumer.Close() no longer races an active ReadMessage and can't trip restartConsumerAsync() into recreating the consumer we just suspended.

The heartbeat's stopChannel case only sets paused=false and breaks — no defensive Resume, since the caller is about to close the consumer anyway.

[jitsu-code-review]

Reviewed the changes in bulker/bulkerapp/app/abstract_batch_consumer.go, focusing on pause/resume/suspend and restart interactions.

The separate suspend stop-signal path and async restart guard address the previously reported race/timeout behavior, and I did not find additional actionable correctness or security issues in this patch.