[PubSub Kafka] retry-blocked ConsumeClaim can't respond to rebalances, causing a self-sustaining single-message oscillation between consumers

## Summary

`pubsub.kafka`'s `ConsumeClaim` processes/retries each message inline, and the same goroutine is responsible for reacting to `session.Context().Done()`. If a consumer-group rebalance is signaled while a member is retrying a message — whether via component-level `consumeRetryEnabled` or only the Dapr runtime's own inbound resiliency policy — that member can't respond to the rebalance promptly. This produces `read tcp ...: i/o timeout` and/or `context canceled` failures, and the resulting eviction/rejoin is itself a rebalance, which hands the still-unacknowledged message to the *other* group member. If that message keeps failing, two members trade it back and forth indefinitely, with no further external trigger required.

We saw this in production (a permanently-failing message due to an unrelated upstream permission error bounced between two consumers of the same app for ~3 hours, surviving a full pod replacement). We have since reproduced the mechanism with a much cleaner, single-message repro: **one health-check blip of ~9.5 seconds is enough to start an indefinite, self-sustaining oscillation of a single message between two consumers, with no further manual intervention.**

**The big problem here is that running many pods to create a HA set-up actually causes a partition block here, which can only be resolved by temporarily stopping one of the pods. The bounded resiliency policy should be allowing the poison message to be dropped, but this doesn't occur, so the partition remains blocked**

## Environment

- `github.com/IBM/sarama`: v1.45.2
- `github.com/dapr/kit`: v0.15.3
- `github.com/cenkalti/backoff/v4`: v4.3.0
- daprd: 1.17.10

## Reproduction

Repro project: https://github.com/olitomlinson/dapr-workflow-examples/tree/kafka-offset-bug (`compose-1-2.yml`) — two app+sidecar pairs (`workflow-app-a`/`workflow-dapr-a`, `workflow-app-b`/`workflow-dapr-b`), same `app-id`, same `pubsub.kafka` component/consumer group, both subscribed to the same topic (single partition).

Component config (note `consumeRetryEnabled: "false"` — the blocking comes entirely from the runtime's own resiliency policy, see Root Cause):

```yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: kafka-pubsub
spec:
type: pubsub.kafka
version: v1
metadata:
- name: brokers
value: "kafka:29092"
- name: consumerGroup
value: "{namespace}"
- name: consumeRetryEnabled
value: "false"
```

```yaml
apiVersion: dapr.io/v1alpha1
kind: Resiliency
metadata:
name: default-pubsub-retry-policy
spec:
policies:
retries:
pubsubRetry:
duration: 40s
maxRetries: 5
policy: constant
targets:
components:
kafka-pubsub:
inbound:
retry: pubsubRetry
```

Both sidecars run with app health probing enabled (`--app-health-check-path /health`, `--app-health-probe-interval 3`, `--app-health-threshold 2`).

Steps:
1. Start both app+sidecar pairs, both in the same consumer group, both subscribed to the same topic/partition.
2. Send one message that the app's pubsub subscription handler will always fail (return 500) for.
3. Briefly flip one app's `/health` endpoint to 500 for a few seconds, then back to 200.
4. Do nothing else. Watch both sidecars' logs.

## Timeline of a single-message repro run (2026-09-11, 15:20-15:36 UTC)

### Phase 0 - baseline, before any toggle

A single message is published to the `monitor-workflow` topic as a manual step. The subscription handler in the test app is hardcoded to always return HTTP 500 for this message, so Dapr classifies it as retriable and falls into the `pubsubRetry` resiliency policy. On its own this is only a *bounded* retry loop (`maxRetries: 5`, ~200s) that would exhaust and stop - it does not yet explain the indefinite oscillation. The permanent, self-sustaining loop only begins once the app health check is manually forced to 500 (Phase 1 below), triggering a rebalance while the message is mid-retry.

| Time | Source | Event |
|---|---|---|
| 15:20:14.462 | dapr-a | `retriable error returned from app while processing pub/sub event 0-10` (offset 28) - A is already alone-retrying this message every ~40s on its own resiliency cadence; health is fine |
| 15:20:54.467 | dapr-a | Same retry, 40s later - confirms A's baseline retry interval with no rebalance involved yet |

### Phase 1 - the trigger (the only manual action in this run)

Manually forcing the app health check to fail is not just "the app looks unhealthy" - it directly triggers daprd to tear down and rebuild the Kafka consumer group membership for that sidecar. In `appHealthChanged` (`pkg/runtime/runtime.go`), an unhealthy transition calls `a.processor.Subscriber().StopAppSubscriptions()`, which (`pkg/runtime/processor/subscriber/subscriber.go`) calls `sub.Stop()` on every active subscription - cancelling the subscription's context and causing the `pubsub.kafka` component to exit its consume loop and **leave the consumer group outright**. When health recovers, the healthy-transition branch calls `StartAppSubscriptions()`, which re-subscribes and **rejoins** the group. Each of these is a real, deliberate join/leave event that the Kafka coordinator observes immediately - not a passive heartbeat timeout - so toggling the health check is a reliable, on-demand way to force an underlying rebalance of the whole group (and, with the `range` strategy, an eager/stop-the-world one).

| Time | Source | Event |
|---|---|---|
| 15:20:29.096 | app-b | Health endpoint status code **changed to 500** |
| 15:20:33.164 | dapr-b | App entered **un-healthy** status -> `StopAppSubscriptions()` runs, dapr-b leaves the consumer group |
| 15:20:38.571 | app-b | Health endpoint status code **changed to 200** |
| 15:20:39.166 | dapr-b | App entered **healthy** status -> `StartAppSubscriptions()` runs, dapr-b rejoins the consumer group |

A ~9.5s health blip on B, which forces one leave + one rejoin of the consumer group. Nothing else is touched manually for the remaining 16+ minutes below.

### Phase 2 - the loop becomes self-sustaining

From here on, the same 3-step pattern repeats indefinitely with no further manual action:

1. **Whichever sidecar currently holds the message retries it against the resiliency policy, then gives up** (retries exhausted) and logs `Error processing Kafka message: .../0/28 ... Error: retriable error occurred`.
2. **~30.1s later, that same sidecar logs `read tcp ...: i/o timeout`** while trying to keep consuming - this is Sarama discovering, after the fact, that its partition assignment was pulled during a rebalance it never got to participate in cleanly (it was busy retrying instead of watching for the rebalance signal - see Root Cause below).
3. **The message has already been silently reassigned to the *other* sidecar**, which starts retrying it from scratch - becoming the new "current holder" for the next cycle.

| Cycle | Held by | Gave up (offset 28) | i/o timeout confirms reassignment (+30.1s) | How long it held the message |
|---|---|---|---|---|
| 1 | dapr-a | 15:23:34.488 | 15:24:04.598 | 3m20s (from first pickup at 15:20:14, Phase 0) |
| 2 | dapr-b | 15:24:59.354 | 15:25:29.465 | 1m25s (since cycle 1's give-up) |
| 3 | dapr-a | 15:27:54.743 | 15:28:24.860 | 2m55s |
| 4 | dapr-b | 15:29:19.630 | 15:29:49.737 | 1m25s |
| 5 | dapr-a | 15:32:15.009 | 15:32:45.117 | 2m56s |
| 6 | dapr-b | 15:33:39.885 | 15:34:09.989 | 1m24s |
| 7 | dapr-a | 15:36:35.283 | *(capture ended before this one logged)* | 2m56s |

Every "give-up" line is identical apart from timestamp/host:
```
Error processing Kafka message: defaultmonitor-workflow/0/28 [key=MTUyN2FiZWYtZDJhNi00OWIzLWJkMDgtNjRkZTE2NWI1ODY5].
Error: retriable error occurred: retriable error returned from app while processing pub/sub event 0-10,
topic: monitor-workflow, body: monitor response is : 500. status code returned: 500.
```
Every i/o-timeout line is identical apart from timestamp/local port:
```
Error consuming [defaultmonitor-workflow]. Retrying...: read tcp :->:29092: i/o timeout
```

Two things stand out: the give-up -> i/o-timeout gap is ~30.1s every single time (sub-10ms variance across all 6 occurrences), and the hold duration alternates cleanly between ~2m55s (dapr-a) and ~1m25s (dapr-b) - a stable, repeating ~4m20s cycle rather than random noise, consistent with the fixed `duration: 40s` / `maxRetries: 5` retry policy plus whatever session/heartbeat timeout constant produces the 30.1s delay.

## Root cause

**1. `ConsumeClaim` processes/retries a message inline, inside the same `select` branch that's supposed to watch for rebalance:**

`common/component/kafka/consumer.go`:

```go
func (consumer *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
b := consumer.k.backOffConfig.NewBackOffWithContext(session.Context())
...
for {
select {
// Should return when `session.Context()` is done.
// If not, will raise `ErrRebalanceInProgress` or `read tcp :: i/o timeout` when kafka rebalance. see:
// https://github.com/IBM/sarama/issues/1192
case <-session.Context().Done():
return nil
case message, ok := <-claim.Messages():
if !ok {
return nil
}
if consumer.k.consumeRetryEnabled {
if err := retry.NotifyRecover(func() error {
return consumer.doCallback(session, message)
}, b, ...); err != nil {
...
}
} else {
err := consumer.doCallback(session, message) // <-- still blocks here even with consumeRetryEnabled=false
if err != nil {
consumer.k.logger.Errorf("Error processing Kafka message: %s/%d/%d [key=%s]. Error: %v.", ...)
}
}
}
}
}
```

This repro has `consumeRetryEnabled: "false"`, so the component-level retry loop (`retry.NotifyRecover`) is not involved. The blocking happens purely inside `doCallback` -> `handlerConfig.Handler(session.Context(), &event)`, which invokes the Dapr runtime's own inbound-resiliency-wrapped delivery to the app. That single call can itself take several minutes (5 retries x up to 40s, per our `pubsubRetry` policy) before returning, and for that entire time `ConsumeClaim` is not back at the `select`, so `session.Context().Done()` goes unnoticed regardless of `consumeRetryEnabled`.

**2. The Dapr runtime's resiliency runner is the source of the multi-attempt blocking:**

`dapr/pkg/resiliency/policy.go`, `NewRunnerWithOptions`:

```go
b := def.r.NewBackOffWithContext(ctx)
...
return retry.NotifyRecoverWithData(
func() (T, error) {
...
rRes, rErr := operation(opCtx)
...
},
b,
func(opErr error, d time.Duration) {
...
def.log.Warnf("Error processing operation %s. Retrying in %v...", def.name, d)
},
...
)
```

Each attempt inside `operation(opCtx)` is a full HTTP round-trip to the app; the retry loop itself only checks context cancellation *between* attempts.

**3. `cenkalti/backoff/v4`'s retry loop only checks context between attempts, never during one:**

`retry.go`, `doRetryNotify`:

```go
func doRetryNotify[T any](operation OperationWithData[T], b BackOff, notify Notify, t Timer) (T, error) {
...
for {
res, err = operation() // blocks here; no ctx check until this returns
if err == nil {
return res, nil
}
...
t.Start(next)
select {
case <-ctx.Done(): // ctx is only checked here, between attempts
return res, ctx.Err()
case <-t.C():
}
}
}
```

Even though the backoff is constructed with `NewBackOffWithContext(ctx)`, that only lets it abort the *sleep* between attempts early. It cannot interrupt an attempt already in flight. Combined with (1), `ConsumeClaim` can be unresponsive to `session.Context().Done()` for as long as a single delivery attempt takes.

**4. On the Sarama side, the coordinator's rebalance signal cancels the session context, but nothing forces `ConsumeClaim` to notice quickly:**

`IBM/sarama@v1.45.2/consumer_group.go`, `heartbeatLoop`:

```go
switch resp.Err {
case ErrNoError:
retries = s.parent.config.Metadata.Retry.Max
case ErrRebalanceInProgress:
retries = s.parent.config.Metadata.Retry.Max
s.cancel() // session.Context() cancelled here
case ErrUnknownMemberId, ErrIllegalGeneration:
return // session torn down; member must fully rejoin
...
}
```

Heartbeats run on their own goroutine (`go sess.heartbeatLoop()`), independent of `ConsumeClaim`, so they are never blocked - the broker never thinks the consumer is dead via a missed heartbeat. The failure path is: the coordinator signals `ErrRebalanceInProgress`, Sarama cancels the session context, but `ConsumeClaim` - still blocked per (1)-(3) - doesn't notice for as long as its current attempt takes. This produces exactly the `read tcp ...: i/o timeout` errors seen above, at a strikingly consistent ~30.1s offset from the give-up.

## Why this becomes a self-sustaining loop, not a one-off

Once one consumer is caught mid-retry when a rebalance starts, it typically has to leave and fully rejoin the group (`ErrUnknownMemberId`/`ErrIllegalGeneration` -> `return`). Rejoining is itself a join event, triggering *another* rebalance. That rebalance hands the still-unacknowledged message's partition to the other member, which immediately retries it, gets caught the same way, and hands it back. With only two members in the group, this alternates indefinitely with no further external trigger required - as demonstrated by the 16+ minute, single-message, single-toggle repro above.

## Impact

- A single permanently-failing message (e.g. due to an unrelated downstream/upstream error misclassified as retriable) can degrade an entire consumer group's throughput indefinitely, not just delivery of that one message.
- Trivial to trigger: a health-check blip lasting under 10 seconds is sufficient to start an indefinite oscillation.
- Hard to diagnose from logs alone: no explicit "rebalance" log line at the point of impact, only generic `i/o timeout` / `context canceled` / `retriable error` lines that look like transient network issues.
- Affects any deployment with `consumeRetryEnabled: true` *or* an inbound resiliency retry policy on the component - i.e. essentially any production Dapr Kafka pubsub setup using retries, which is the recommended configuration.

## Suggested fix directions

1. Run message delivery (including any retry/backoff) off the `ConsumeClaim` goroutine - e.g. hand off to a worker and keep the main loop selecting on `session.Context().Done()` so it can react immediately regardless of how long an in-flight delivery takes.
2. At minimum, make the per-attempt operation passed into the resiliency runner genuinely cancellable mid-flight (not just between attempts), so a cancelled `session.Context()` interrupts it within a bounded, short time rather than up to a full backoff interval.
3. Log distinctly when `ConsumeClaim` observes `session.Context().Done()` significantly later than it was actually cancelled, so operators can detect this condition instead of seeing only generic timeout/cancellation noise.
4. Consider defaulting to (or documenting) `CooperativeStickyAssignor` for this component, so a single member's temporary unresponsiveness only affects the partitions that actually need to move, rather than forcing a stop-the-world reassignment of the whole group.

## References

- `dapr/components-contrib`: `common/component/kafka/consumer.go` (`ConsumeClaim`)
- `dapr/dapr`: `pkg/resiliency/policy.go` (`NewRunnerWithOptions`)
- `cenkalti/backoff/v4@v4.3.0`: `retry.go` (`doRetryNotify`)
- `IBM/sarama@v1.45.2`: `consumer_group.go` (`heartbeatLoop`)
- Referenced upstream issue in the existing code comment: https://github.com/IBM/sarama/issues/1192

## Note on `consumeRetryEnabled` (this repro used `false`; the component default is `true`)

The repro above uses `consumeRetryEnabled: "false"` to isolate the runtime-level resiliency retry cleanly. It's worth noting that `pubsub.kafka` actually **defaults to `consumeRetryEnabled: true`** (`pubsub/kafka/kafka.go`), so most real deployments have both retry layers active, nested. We believe this makes the bug harder to diagnose, not easier to avoid.

With `consumeRetryEnabled: true`, `ConsumeClaim` wraps `doCallback` in a second, outer retry loop (`consumer.go:87`) using the component's own `backOffConfig`, separate from the Dapr runtime's inbound `pubsubRetry` policy that already runs *inside* `doCallback`. The component's backoff defaults (`dapr/kit/retry.DefaultConfig()`) are:

```go
Policy: PolicyConstant,
Duration: 5 * time.Second,
MaxRetries: -1, // -1 means no cap - retries forever
```

`NewBackOff()` only applies `backoff.WithMaxRetries(...)` when `MaxRetries >= 0`, so with the default `-1` the outer loop retries indefinitely, waking every 5s to call `doCallback` again - and each call re-enters the ~200s inner cycle (40s x 5 retries) before it can fail. There is no point at which `doCallback` returns an error up to `ConsumeClaim`'s non-retry branch, because that branch doesn't run at all when `consumeRetryEnabled: true`.

Practical consequences we'd expect, compared to what we captured with it disabled:

1. **No more clean "give up -> return to `select`" checkpoint.** In our repro, the pod finishes its ~200s inner cycle, fails, logs `Error processing Kafka message: ... retriable error occurred`, and immediately loops back to `ConsumeClaim`'s outer `select` - which is exactly the moment it can notice `session.Context().Done()`. With retries enabled, that return never happens on its own while the message keeps failing.
2. **The window where `session.Context().Done()` can be noticed shrinks drastically.** `cenkalti/backoff`'s `doRetryNotify` only checks `ctx.Done()` between attempts of the *outer* retry - a ~5s window once per ~200s cycle, versus what we observed with retries disabled where the pod naturally revisited `select` every cycle. That's roughly a 2.5% duty cycle for detecting a rebalance signal, rather than effectively every cycle.
3. **We'd expect longer, less predictable hold times and a harder failure mode**, rather than the clean, bounded ~1m25s/~2m55s alternation captured above - a pod could hold the poison message through many rebalance signals purely by missing the narrow 5s window each time, and might eventually be evicted via a broker-side `session.timeout.ms` rather than the orderly context-cancellation path shown here.

We haven't captured log evidence for this variant yet, but wanted to flag it: `consumeRetryEnabled: false` doesn't fix the underlying issue, it just bounds each episode to ~200s and gives the mechanism a reliable, frequent opportunity to interact cleanly with a rebalance. The default (`true`) removes that opportunity almost entirely.

Start with ConsumeClaim in common/component/kafka/consumer.go and trace how session.Context().Done() is handled while doCallback is retrying. Then read NewRunnerWithOptions in dapr/pkg/resiliency/policy.go and heartbeatLoop in Sarama's consumer_group.go. Run the compose-1-2.yml reproduction and verify that a rebalance interrupts the blocked consumer without causing the message to oscillate between consumers.

Written by the indexing model from the issue text.