Fatal "Static consumer fenced by other consumer with same group.instance.id" on consumer restart, causes permanent silent stall with no retry

[roman-pepper]

Summary

AbstractBatchConsumer.restartConsumer() can trigger Kafka's static-membership fencing against itself, because it starts a replacement consumer with the same group.instance.id before confirming the previous consumer has actually left the group. When this race is lost, bulker treats the resulting fencing error as fatal and never retries -- the consumer for that topic goes permanently silent (no further log lines at all) until the process is manually restarted.

This caused a multi-hour to multi-day silent ingestion outage for us on two separate occasions, since nothing else in the system detects "consumer stopped producing any log output at all" as distinct from "consumer idle because there's nothing to consume."

Where

• bulker/bulkerapp/app/abstract_consumer.go, GetInstanceId():

  // range partitioner assigner distributes partitions between consumers in alphabetical order
  // since bulker topics mostly have only 1 partition – instance with the lowest instanceId will be assigned for all topic.
  // we use first letters of hash of 'topicId + instanceId' as a beginning of 'group.instance.id'
  // so for each topic the first instance will be different
  // while keeping consistency between restarts (if instanceId is the same)
  firstByte := md5.Sum([]byte(ac.topicId + ac.config.InstanceId))[0]
  return fmt.Sprintf("%x-%s", firstByte, ac.config.InstanceId)

  Since config.InstanceId is fixed for the process lifetime, every consumer recreated for the same topic within the same process reuses the exact same group.instance.id by design.

• bulker/bulkerapp/app/abstract_batch_consumer.go, restartConsumer():

  bc.Infof("Restarting consumer")
  go func(c *kafka.Consumer) {
      e1 := c.Unsubscribe()
      e2 := c.Close()
      bc.Infof("Previous consumer closed: %v %v", e1, e2)
  }(bc.consumer.Load())
  
  ticker := time.NewTicker(time.Duration(bc.config.KafkaSessionTimeoutMs+15000) * time.Millisecond)
  ...
  case <-ticker.C:
      ...
      _, err := bc.initConsumer(true)

  The old consumer's Close() runs in an unawaited goroutine. The code waits KafkaSessionTimeoutMs + 15s before creating the replacement (presumably to give the old member time to leave the group before the new one claims the same static ID), but there's no actual confirmation that the close completed -- just a fixed timer. Under sustained heavy load, Close() can apparently take longer than that window, and the new consumer joins with the same group.instance.id while the old one hasn't fully left, producing:

  Fatal error: Fatal consumer error: Broker: Static consumer fenced by other consumer with same group.instance.id
  

  This is logged via the normal error path but nothing in the code retries it -- the consumer just stops.

Impact

Once this fires, the affected topic's consumer produces zero further log output of any kind -- not even the periodic degradation/retry warnings -- until the process is externally restarted. For a high-volume topic, this is silent, total data-ingestion loss for that destination with no automatic recovery and no distinguishable signal from "topic is just idle."

Observed conditions

We saw this trigger repeatedly during sustained backlog catch-up (millions of queued messages after an unrelated infra restart), including:

• Once during a normal ECS rolling deployment (brief two-task overlap, each with the same static ID from a shared image/config -- expected collision)
• Twice with only one instance running, no deployment overlap at all -- the race triggered purely from bulker's own internal consumer-recreation cycle, roughly ~5-6 minutes into a long-running batch, right as the internal "no successful consume" watchdog fired and tried to restart the consumer

Sample log sequence (one instance, no overlapping tasks):

19:01:54 level=warning msg="[topic] Consumer degradation detected: no successful consume for 5m0s ... Recreating consumer."
19:01:54 level=info msg="[topic] Restarting consumer"
19:01:54 level=info msg="[topic] Previous consumer closed: <nil> <nil>"
19:02:54 level=info msg="[topic] Consumer created: bulkerapp#consumer-7"
19:02:54 level=info msg="[topic] Successfully altered consumer group offsets: ..."
19:02:54 level=info msg="[topic] Consumer created: bulkerapp#consumer-9"
19:02:54 level=error msg="[topic] Consume finished with error: ... Fatal error: Fatal consumer error: Broker: Static consumer fenced by other consumer with same group.instance.id stats: consumed: 1080000 processed: 1080000 ... time: 184.98 s."
(no further log lines for this topic for the next ~17 minutes, until the process was manually restarted)

Note two Consumer created lines within the same second (consumer-7, then consumer-9) -- suggests overlapping restart attempts even within a single process.

Suggested directions

1. Make restartConsumer actually wait for the old consumer's Unsubscribe()/Close() to complete (or for explicit leave-group confirmation) before calling initConsumer(true), rather than relying on a fixed timer with no completion signal.
2. Treat the "Static consumer fenced by other consumer with same group.instance.id" error as retryable with backoff rather than fatal-with-no-retry, since it's a transient self-inflicted race rather than a genuine unrecoverable state.
3. Consider logging a distinct, loud signal when a consumer's goroutine exits after a fatal error (vs. going silent), so downstream monitoring can tell "stopped forever" apart from "just idle."

Happy to provide more log context or test against a patched build if useful.

[sohammehta06]

Thanks for this, it's an unusually good report.

The two pieces you've put together look right to me: config.InstanceId being fixed for the process lifetime means every consumer recreated for the same topic reuses the same group.instance.id, and restartConsumer() closing the previous consumer in a goroutine means the replacement can come up before the old member has actually left the group. Treating the resulting fencing error as fatal with no retry, and then emitting no further log lines at all, is a genuinely bad combination, since a silent consumer is indistinguishable from one that's idle because there's nothing to consume.

I'm on the growth side rather than engineering, so i'm not going to pretend to confirm a fix here, but i'm flagging it internally and i'll make sure it gets a proper look rather than sitting.

Two things that would help size it:

• roughly what event volume and how many topics were you running when it hit?
• are you on the community helm chart (stafftastic/jitsu-chart) or your own deployment?

And if it's useful, i'd happily get you on a call with our CTO. Either way, thanks for digging this far into it.

[roman-pepper]

@sohammehta06 Thanks for the quick response, and for looking into it further.

Volume: roughly 3.5-5M events/day normally, essentially all of it flowing through a single high-volume workspace's topic (a second workspace topic on the same deployment runs at a much smaller trickle, maybe a few hundred events/batch). Both of those, plus our internal destination-messages-retry topic, are single-partition. We also produce to (but don't consume) a destination-messages topic that has 16 partitions, if that's relevant. During the incident, this consumer was also catching up a backlog of 8-9M+ queued events, so the batches it was working through were much larger than normal, which is likely what pushed batch completion times past the 5-minute watchdog often enough to trigger this.

Deployment: our own, not the community Helm chart. We run bulker as its own container (jitsu-service:bulker-latest, mirrored into our own ECR) on AWS ECS Fargate, not Kubernetes.

Appreciate you flagging it internally rather than letting it sit. I'll pass on the call for now, but happy to keep answering questions here if that's useful for scoping the fix.

[absorbb]

@roman-pepper
I believe this issue must be fixed by this PR #1496
that completely refactors consumer's lifecycle implementation
due too significance of change we need more time to review and test this PR