[Kafka pubsub] Offer an opt-in per-topic consumer group mode to improve fault isolation between subscribed topics

**Summary**

Today, when an application subscribes to multiple topics via the Kafka pubsub component, all topics share a single `sarama.ConsumerGroup` instance, created once from the component's consumerGroup metadata field and reused for every `Subscribe()` call:

```
// common/component/kafka/clients.go
cg, err := sarama.NewConsumerGroup(k.brokers, k.consumerGroup, k.config)
```

Every subscribed topic is merged into one subscribeTopics map and passed as a single slice into one `Consume(ctx, topics, consumer)` call:

```
// common/component/kafka/subscriber.go
topics := k.subscribeTopics.TopicList()
...
err = clients.consumerGroup.Consume(ctx, topics, consumer)
```

This means every subscribed topic shares one broker connection (or connection pool) and one group-coordinator session. **We'd like to propose an opt-in second mode where the component instead creates one consumer group per subscribed topic — without changing or removing today's default (shared-group) behavior.** See the Appendix for the incident that prompted this.

**Motivation**

Independent of any specific failure mechanism, the current architecture means: any connection-level or session-level fault affecting the shared group — a broker restart, a network blip, an idle-connection timeout somewhere in the path, a slow/stuck partition consumer — has a direct path to affect every other topic multiplexed onto that same connection and session, purely because they're grouped together. A per-topic consumer group would remove that specific coupling by construction, regardless of which exact fault triggers it.

**Proposal**

Add an opt-in mode, configurable via component metadata (e.g. consumerGroupStrategy: shared | per-topic, defaulting to shared to preserve current behavior), where:

- Each subscribed topic gets its own `sarama.ConsumerGroup` instance and its own group ID (e.g. derived as `-`, or via an explicit per-topic override).
- Each topic's `Consume()` loop, retry/backoff, and lifecycle (subscriber.go's `consume()`) runs independently.
- Existing single-group behavior remains the default; nothing changes for users who don't opt in.

**Trade-offs we're aware of**

**Potential benefits:**
- A connection or session fault on one topic's group has no path to affect another topic's group — they're fully independent connections and coordinator sessions.
- A rebalance on one topic (scaling, restart, slow consumer) doesn't block or delay consumption on other topics sharing the same subscription today.
- Diagnostics become topic-scoped by construction — logs/metrics naturally identify which topic's group is affected, instead of a shared log line listing every topic together.

**Potential costs:**
- N topics means up to N separate broker connections and N separate group-coordinator sessions per app instance, instead of one — more connection/heartbeat overhead, which could matter at fleet scale (many app instances × many topics).
- Consumer group ID design needs care for backward compatibility: existing deployments have committed offsets under today's single group ID. Migrating a given topic to its own group would need either a defined ID derivation scheme users can reason about, or an accepted offset reset — this is why we're proposing it as opt-in rather than a change to the default.
- More independent moving parts to reason about operationally (N consume loops instead of 1) — arguably better for isolation, but it is more surface area.
- This does not, on its own, address the underlying per-partition behavior in ConsumeClaim (a slow/failing message retried synchronously in the same goroutine that reads
`claim.Messages()`). It narrows the blast radius from "every topic in the subscription" to "the topic in question," but a single topic with a stuck partition could still stall itself under this mode.

**Ask**

**We'd like feedback from maintainers on:**
1. Whether this is architecturally reasonable to add as an opt-in mode without disrupting the existing shared-group path.
2. Preferred approach for per-topic group ID derivation/configuration.
3. Whether this is something we could help prototype, given interest.

---
**Appendix A: incident that prompted this proposal**

We hit an incident where a Kafka client connection error (`write tcp ...: broken pipe`) appeared on a shared consumer group covering five subscribed topics. The log line for that error names all five topics together, because they're all arguments to the same `Consume()` call:

`Error consuming [group topic-a topic-b topic-c topic-d topic-e]. Retrying...: write tcp ->:9092: write: broken pipe`

Immediately after that error, we observed a burst of previously-undelivered messages land across multiple of those topics at once, consistent with the shared group needing to fully reconnect and rejoin before any of them could resume. Separately, broker-side logs showed the shared group's session reset to an "Empty" state with no clean `LeaveGroup` — i.e., the session was lost at the connection level, not by application choice.

We were not able to fully pin down the exact internal mechanism that caused one specific topic to stop being delivered for an extended period while this was happening. We suspect it may relate to a slow-processing partition on one topic combined with the shared broker-connection-level fetch cycle, but the relevant client-library debug logging wasn't enabled at the time of the incident, and broker-side group-coordinator logs don't capture that layer of detail — so we want to be explicit that we could not fully confirm the exact mechanism from available telemetry. What we could confirm is the architectural fact above: all five topics were coupled through one connection and one session, and a fault on that shared session had a plausible path to affect all of them simultaneously.

---
**Appendix B: precedent for per-concern consumer group isolation in production**

To gauge whether maintaining several independent Kafka consumer groups within a single running service is a reasonable established pattern (as opposed to an edge case), we looked at an internal microservices codebase we operate that makes heavy use of Kafka (outside of dapr) across a number of services. We found that the general convention there is the **inverse** of Dapr's current default: rather than multiplexing multiple topics/concerns onto one shared consumer group per service, individual services deliberately instantiate multiple independent consumer groups — each with its own group ID, its own client ID, and its own broker connection — within a single process, one per logical concern.

**Two examples stood out:**

- One orchestration service runs six independent consumer-group identities concurrently in a single process: several plain Kafka consumers (each with a distinct group.id and client.id) alongside multiple embedded Kafka Streams applications (each with its own application.id), all started concurrently as sibling sub-components of one service. Each sub-component's consumer is tuned independently (partition-assignment strategy, max fetch bytes, low-latency settings), which is only possible because each maintains its own connection rather than sharing one.
- Another scheduling service runs four independent consumer-group identities in one process for similar reasons — splitting distinct workloads (e.g., a queue vs. a schedule feed) onto separate consumers with distinct client IDs, even though both run in the same pod and point at the same broker list.

In both cases, every consumer within the process points at the same Kafka cluster (bootstrap.servers) — the separation is purely at the consumer-group/connection level, not a multi-cluster routing setup. There is no shared connection-pool abstraction anywhere in this codebase; the convention is simply "a new independent concern gets its own consumer object with its own group ID," which is structurally the same shape as the opt-in per-topic mode proposed above (same broker(s), dedicated group/connection per concern instead of one shared multiplexed connection). The team's own internal architecture documentation independently corroborates this, listing distinct consumer-group IDs per logical sub-component rather than one shared ID per process.

We think this is useful evidence that maintaining multiple independent Kafka connections from a single service process is a normal, deliberate, and tractable pattern in practice — not an unusual or discouraged one — and that Dapr offering it as an opt-in mode would be consistent with how teams already choose to use Kafka when given the choice.

The net gain here (other than choice) is that Dapr could chose of offer a more durable & resilient option for those that want it, and _could_ eventually become the future default choice.

Start with common/component/kafka/clients.go and common/component/kafka/subscriber.go, especially ConsumerGroup creation, Subscribe(), Consume(), and consume(). Read the existing metadata and lifecycle handling before proposing the opt-in configuration. Done means maintainers agree on the strategy and group-ID behavior while preserving shared-group defaults and isolating per-topic consumption.

Written by the indexing model from the issue text.