GH-4190 Add support to KIP-848

[lillo42]

Description

Add support for KIP-848 — Kafka's new consumer group protocol — alongside the existing classic protocol.

• Introduces an IGroupProtocol strategy applied to the ConsumerConfig at consumer creation:
  • ClassicGroupProtocol (default): carries SessionTimeout, HeartbeatInterval, and PartitionAssignmentStrategy. When no protocol is supplied, the consumer back-fills any null properties from its constructor parameters (session timeout default 10 s, assignment strategy default RoundRobin), preserving current behavior. The back-fill mutates the instance in place, so an instance must not be shared across subscriptions (documented on the type and on KafkaSubscription.GroupProtocol).
  • ConsumerGroupProtocol (KIP-848): carries GroupRemoteAssignor and GroupInstanceId, and deliberately does not set session.timeout.ms, heartbeat.interval.ms, or partition.assignment.strategy, which librdkafka rejects when group.protocol=consumer.
• KafkaSubscription.GroupProtocol selects the protocol; the legacy SessionTimeout and PartitionAssignmentStrategy properties on KafkaSubscription are obsoleted (with #pragma guards at call sites) in favor of ClassicGroupProtocol.
• XML documentation covers the back-fill precedence and no-sharing contract, the static-membership caveat (GroupInstanceId must be unique per consumer instance — only set it with a single performer, or assign unique ids via configHook), and the weaker infrastructure-validation guarantees of the consumer protocol (broker-driven membership completes asynchronously, so OnMissingChannel.Validate cannot reliably detect missing topics; prefer Create or provision topics out-of-band).

Test infrastructure

• Migrated the Kafka test broker to KRaft (apache/kafka, no ZooKeeper) in docker-compose-kafka.yaml and the CI workflow; images pinned to explicit versions (apache/kafka:4.0.2, cp-schema-registry:8.0.6, cp-enterprise-control-center:7.9.8).
• Added a generated "Consumer" messaging-gateway test suite (KIP-848 protocol) alongside the renamed "Classic" suite; infrastructure-validation tests are excluded for the consumer protocol for the reason above.
• Added broker-free unit tests over ClassicGroupProtocol.Apply / ConsumerGroupProtocol.Apply that lock in the KIP-848 config contract (consumer protocol never sets the classic-only keys).
• KafkaMessageAssertion now compares timestamps normalized to UTC seconds (the previous string comparison was timezone-sensitive).

Related Issues

#4190

Type of Change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the Contributing Guide
• I have checked the documentation for relevant guidance
• I have added/updated XML documentation for any public API changes
• I have added/updated tests as appropriate
• My changes follow the existing code style and conventions

Additional Notes

Addresses the automated review feedback: documented the IGroupProtocol no-sharing contract and the GroupInstanceId static-membership caveat, added unit tests over the Apply mappings, pinned the Docker images, and fixed the cosmetic issues (trailing whitespace/newline, double space).

[claude]

Code Review — GH-4190 Add support to KIP-484 (Kafka consumer group protocol)

Thanks for this! Exposing GroupProtocol, GroupRemoteAssignor, and GroupInstanceId on KafkaSubscription and threading them through the factory into ConsumerConfig is clean and well-documented, and the doc-comment/typo cleanups in KafkaMessageConsumer are a nice touch. A few things worth addressing before merge.

🔴 CI/test infrastructure likely cannot run the new Consumer protocol tests

KafkaConsumerMessageGatewayProvider creates subscriptions with GroupProtocol = GroupProtocol.Consumer (KIP-848 next-gen consumer group protocol). That protocol requires a broker in KRaft mode with the new group coordinator enabled (group.coordinator.rebalance.protocols=...,consumer).

However, both docker-compose-kafka.yaml and .github/workflows/ci.yml run confluentinc/cp-enterprise-kafka:latest in ZooKeeper mode (KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181), where the new consumer protocol is not available. As it stands, the new Consumer/... integration tests will very likely fail (or the broker will reject group.protocol=consumer) in CI. Please confirm they pass against the CI broker; if not, the compose/CI Kafka service needs KRaft + the new-coordinator enablement.

🟠 GroupInstanceId is silently ignored under the (default) Classic protocol

if (groupProtocol == GroupProtocol.Classic)
{
    _consumerConfig.PartitionAssignmentStrategy = partitionAssignmentStrategy;
    _consumerConfig.SessionTimeoutMs = ...;
}
else if (groupProtocol == GroupProtocol.Consumer)
{
    _consumerConfig.GroupProtocol = groupProtocol;
    _consumerConfig.GroupRemoteAssignor = groupRemoteAssignor;
    _consumerConfig.GroupInstanceId = groupInstanceId;   // only applied here
}

Static membership (group.instance.id, KIP-345) is a Classic-protocol feature, but it is only applied in the Consumer branch. A user who sets GroupInstanceId while leaving GroupProtocol at its default (Classic) will have it silently dropped. The XML doc ("Set this to enable static membership and reduce rebalances for restarting consumers") gives no hint of a protocol restriction — behavior and docs disagree. Consider applying GroupInstanceId in both branches, or documenting the restriction.

🟠 Mismatched settings are dropped without warning

If GroupRemoteAssignor/GroupInstanceId are set with Classic (the default), or PartitionAssignmentStrategy is set with Consumer, the value is silently ignored. A ConfigurationException (consistent with the existing groupId/routingKey validation in this constructor) or at least a doc note would prevent surprising misconfiguration.

🟡 Behavior change: default partitionAssignmentStrategy on the constructor

- PartitionAssignmentStrategy partitionAssignmentStrategy = PartitionAssignmentStrategy.RoundRobin,
+ PartitionAssignmentStrategy? partitionAssignmentStrategy = null,

For callers constructing KafkaMessageConsumer directly (not via KafkaSubscription, which still defaults to RoundRobin), a null value flows straight into ConsumerConfig.PartitionAssignmentStrategy, so librdkafka falls back to its own default (range,roundrobin) rather than Brighter previous RoundRobin. That is a silent default change for direct constructor users. CLAUDE.md asks not to change defaults beyond what was requested — consider keeping the RoundRobin default (still skipping the assignment under Consumer) or noting this as intentional in the PR/changelog.

🟡 Test coverage: no broker-free unit test for the config mapping

The core of this PR is the branching that maps protocol to ConsumerConfig fields, yet the only new tests are broker-dependent integration tests. The constructor already accepts a configHook(_consumerConfig), so a fast unit test could capture the config and assert:

• Classic -> SessionTimeoutMs/PartitionAssignmentStrategy set, GroupProtocol unset;
• Consumer -> GroupProtocol/GroupRemoteAssignor/GroupInstanceId set, SessionTimeoutMs unset.

This locks in the mapping without a KIP-848-capable broker (and sidesteps the CI concern above). Per CLAUDE.md TDD workflow, this branching logic is the natural place for a unit test.

🟢 Minor / nits

• else if(groupProtocol == GroupProtocol.Consumer) — missing space after if.
• If GroupProtocol ever gains a third value, neither branch runs and SessionTimeoutMs/PartitionAssignmentStrategy are left unset; a final else/exception would be defensive (low priority given the enum has two members today).

Overall the API surface and wiring look correct; the main blockers are confirming the CI broker can exercise the Consumer protocol and the GroupInstanceId-under-Classic behavior. Nice work otherwise. 👍

_{🤖 Automated review by Claude Code (Opus 4.8). Please verify findings against a real broker before acting.}

[claude]

Review: GH-4190 Add support to KIP-484

Thanks for this — exposing the new broker-coordinated consumer group protocol (GroupProtocol.Consumer, GroupRemoteAssignor, GroupInstanceId) is a welcome addition, and it's nice to see the CI/compose migrated off ZooKeeper to KRaft along the way. The wiring from KafkaSubscription → KafkaMessageConsumerFactory → KafkaMessageConsumer is clean, and the split of SessionTimeoutMs/PartitionAssignmentStrategy into the Classic-only branch is correct (the new protocol rejects client-side assignment and session timeout). Integration coverage via the duplicated generated suite (Classic + Consumer) is a good approach.

A few things worth considering:

Design / correctness

1. GroupInstanceId is silently dropped for the Classic protocol. In KafkaMessageConsumer the property is only applied inside the GroupProtocol.Consumer branch:

else if (groupProtocol == GroupProtocol.Consumer)
{
    _consumerConfig.GroupProtocol = groupProtocol;
    _consumerConfig.GroupRemoteAssignor = groupRemoteAssignor;
    _consumerConfig.GroupInstanceId = groupInstanceId;
}

Static membership (group.instance.id, KIP-345) is valid with the classic protocol too — arguably that's where most people use it today. As written, a user who sets GroupInstanceId while leaving GroupProtocol at its default (Classic) gets it silently ignored, which contradicts the property's own XML doc ("Set this to enable static membership and reduce rebalances for restarting consumers") that makes no mention of a protocol restriction. Consider applying GroupInstanceId regardless of protocol (set it before/outside the branch), or at minimum document the restriction on the property.

2. Broker-side enablement of the consumer protocol. The new protocol requires the broker to have it enabled. It's GA/default only from Apache Kafka 4.0; earlier versions need group.coordinator.rebalance.protocols=classic,consumer (and, in some versions, group.coordinator.new.enable). The compose/CI use apache/kafka:latest (currently 4.x, so it works today) but set no KAFKA_GROUP_COORDINATOR_REBALANCE_PROTOCOLS. This makes the Consumer-protocol integration tests dependent on latest staying ≥ 4.0. Pinning the image to a concrete 4.x tag (and/or explicitly setting the rebalance-protocols env) would make the suite reproducible and robust against a future latest bump or a rollback.

Consistency / minor

3. New options are only settable via object initializer. GroupProtocol / GroupRemoteAssignor / GroupInstanceId were added as auto-properties but not threaded through either KafkaSubscription constructor, unlike every other option on the class. That's functional (the tests use { GroupProtocol = ... }), but it's inconsistent with the established constructor-parameter pattern and easy to miss when discovering the API.

4. Provider duplication. KafkaConsumerMessageGatewayProvider and KafkaClassicMessageGatewayProvider are ~260 lines that differ only in two using namespaces and the single GroupProtocol = GroupProtocol.Consumer initializer. If the generator model allows it, a shared base with an overridable CreateSubscription would cut the maintenance surface. (Understood if this is dictated by the test generator.)

5. KIP number. The implemented feature — group.protocol=consumer / group.remote.assignor — is the next-gen rebalance protocol from KIP-848, not KIP-484. The linked issue #4190 already uses "484", so this is a pre-existing mislabel rather than something introduced here, but it may be worth correcting the title/issue for future searchability.

Nice touches

• KafkaMessageAssertion.NormalizeToUtcSecond is a real improvement — normalizing to UTC before truncating to the second fixes a latent timezone-offset comparison bug in the old ToString(...) approach.
• The doc-comment cleanups (typo fixes, unacceptableMessageLimitWindow, clarified Reject/Purge remarks) are appreciated.

Nit: docker-compose-kafka.yaml now ends without a trailing newline.

Overall this looks solid and well-tested; #1 (the GroupInstanceId/Classic gap) is the one I'd most want addressed before merge, with #2 (broker enablement / image pinning) close behind for CI stability.

🤖 Generated with Claude Code

[claude]

Claude Code Review — PR #4233 (GH-4190 Add support to KIP-848)

Thanks for this — the migration of the Kafka test infrastructure to KRaft and the addition of the new consumer group protocol is a clean, well-scoped change. Overall this looks good. A few observations, one of which I think is a genuine gap.

🐛 Potential bug — GroupInstanceId is silently ignored under the default (Classic) protocol

In KafkaMessageConsumer the group instance id is only applied inside the GroupProtocol.Consumer branch:

if (groupProtocol == GroupProtocol.Classic)
{
    _consumerConfig.PartitionAssignmentStrategy = partitionAssignmentStrategy;
    _consumerConfig.SessionTimeoutMs = Convert.ToInt32(sessionTimeout.Value.TotalMilliseconds);
}
else if (groupProtocol == GroupProtocol.Consumer)
{
    _consumerConfig.GroupProtocol = groupProtocol;
    _consumerConfig.GroupRemoteAssignor = groupRemoteAssignor;
    _consumerConfig.GroupInstanceId = groupInstanceId;   // only here
}

But the KafkaSubscription.GroupInstanceId doc says:

Set this to enable static membership and reduce rebalances for restarting consumers.

Static membership (group.instance.id, KIP-345) is valid and widely used with the classic protocol as well — arguably that is its most common use today. As written, a user who sets GroupInstanceId while leaving GroupProtocol at its default (Classic) gets a silent no-op with no warning. Consider setting GroupInstanceId for both protocols (it is safe to set regardless of protocol), or at minimum documenting that it only takes effect with GroupProtocol.Consumer.

Conversely, correctly omitting SessionTimeoutMs and PartitionAssignmentStrategy for the Consumer protocol is the right call — both are broker-managed / unsupported client-side under KIP-848, so this avoids "property not supported" errors. 👍

🧹 API consistency — new group settings are property-only, not constructor parameters

GroupProtocol, GroupRemoteAssignor, and GroupInstanceId were added as public settable properties but not as constructor parameters, unlike every other Kafka subscription option (which are all threaded through the constructor). It works via object-initializer syntax (the tests use exactly that), but it is inconsistent with the established KafkaSubscription / KafkaSubscription<T> API surface and easy to miss. Consider adding them to both constructors for parity, or note this if the property-only approach is intentional.

✅ Things I liked

• KRaft migration (ci.yml + docker-compose-kafka.yaml): dropping ZooKeeper is necessary for KIP-848 and the KRaft single-node config looks correct, including the CONTROLLER listener wiring and the PLAINTEXT:// prefix fix for schema-registry's bootstrap.
• KafkaMessageAssertion timestamp normalization: replacing the raw ToString("yyyy-MM-ddTHH:mm:ss") comparison with NormalizeToUtcSecond removes a latent timezone/offset flakiness in the assertion. Nice catch.
• Test coverage: the new Consumer generated suite + KafkaConsumerMessageGatewayProvider exercise the KIP-848 path end-to-end, and test-configuration.json correctly marks infra-validation as unsupported for that variant. The Standard→Classic rename is applied consistently (no stale references remain).

Minor nits

• docker-compose-kafka.yaml: trailing whitespace on image: apache/kafka:latest and a missing trailing newline at EOF.
• Both ci.yml and the compose file pin apache/kafka:latest. That matches the pre-existing style (the old images were :latest too), so no regression — but an unpinned tag makes CI non-reproducible and could break unexpectedly when a new Kafka major ships. Pinning to a concrete version would be more robust.

Test-coverage suggestion (optional)

There is no unit-level assertion that the config wiring itself is correct — e.g. that GroupProtocol.Consumer produces a ConsumerConfig with GroupProtocol set and SessionTimeoutMs/PartitionAssignmentStrategy left unset (and vice-versa for Classic). The integration tests cover the happy path but require a live broker. A small unit test around the config construction would cheaply guard against regressions here (and would have caught the GroupInstanceId gap above).

---

Automated review by Claude Code. Take or leave any point as you see fit — the GroupInstanceId behaviour is the one I would most encourage a second look at.

[iancooper]

Nothing wrong with the way it is, so approved, but I would be interested if we could collate those parameters better.

[claude]

Claude Code Review — PR #4233 (KIP-848 support)

Thanks for this! The IGroupProtocol abstraction is a nice cleanup — replacing the enum + two loose string properties with strategy objects (ClassicGroupProtocol / ConsumerGroupProtocol) reads much better and keeps classic-only vs consumer-only settings from leaking into each other. The groupProtocol ??= new ClassicGroupProtocol { SessionTimeoutMs = sessionTimeout } default preserves the old default nicely. A few things to look at:

Bug — PartitionAssignmentStrategy is no longer applied (behavioral regression)

In the old code the classic branch set two things:

_consumerConfig.PartitionAssignmentStrategy = partitionAssignmentStrategy;
_consumerConfig.SessionTimeoutMs = Convert.ToInt32(sessionTimeout.Value.TotalMilliseconds);

ClassicGroupProtocol.Apply (ClassicGroupProtocol.cs:44-57) only sets GroupProtocol, SessionTimeoutMs, and HeartbeatIntervalMs — PartitionAssignmentStrategy is dropped entirely. Meanwhile the constructor still accepts partitionAssignmentStrategy (KafkaMessageConsumer.cs:125, default RoundRobin) and KafkaSubscription.PartitionAssignmentStrategy still flows into it via the factory — but the value is now dead: it is never written to _consumerConfig.

Impact:

• Existing users who relied on Brighter's RoundRobin default (or who explicitly set a strategy) now silently fall back to librdkafka's default (range,roundrobin), which can change partition assignment / rebalancing behavior.
• The removed cooperative-rebalancing comment/link suggests this setting was deliberate.

Suggested fix: add a PartitionAssignmentStrategy property to ClassicGroupProtocol and apply it in Apply, and seed it into the default (new ClassicGroupProtocol { SessionTimeoutMs = sessionTimeout, PartitionAssignmentStrategy = partitionAssignmentStrategy }). Otherwise the partitionAssignmentStrategy parameter should be removed so it isn't misleadingly accepted-and-ignored.

Breaking API change vs. PR classification

The PR is marked "New feature (non-breaking change)", but this is source-breaking:

• KafkaSubscription.GroupProtocol changed type from GroupProtocol (enum) to IGroupProtocol?.
• KafkaSubscription.GroupRemoteAssignor and .GroupInstanceId public properties were removed.
• The KafkaMessageConsumer public constructor lost three parameters and changed the type of another.

Anyone currently setting GroupProtocol = GroupProtocol.Consumer or those properties will fail to compile. Worth calling out in release notes as breaking, or considering keeping the old members as [Obsolete] for a deprecation cycle.

Minor — naming consistency

ClassicGroupProtocol.SessionTimeoutMs is a TimeSpan? but carries an Ms suffix, while the sibling HeartbeatInterval (also TimeSpan?) does not. The Ms suffix reads as "this is milliseconds" when it is actually a TimeSpan. Consider SessionTimeout / HeartbeatInterval for consistency, since the ms conversion already happens inside Apply.

Test coverage

The new Consumer generated integration tests are a good addition and the Standard->Classic rename is sensible. However, both new protocol classes are only exercised through broker-backed integration tests. A small, fast unit test asserting Apply behavior would guard the regression above cheaply — e.g. ClassicGroupProtocol.Apply sets GroupProtocol.Classic + the expected SessionTimeoutMs/HeartbeatIntervalMs (and, once fixed, PartitionAssignmentStrategy), and ConsumerGroupProtocol.Apply sets GroupProtocol.Consumer + assignor/instance id. Per the repo's TDD guidance, a test for the partition-strategy behavior would have caught the drop.

Nits

• docker-compose-kafka.yaml now uses apache/kafka:latest; consider pinning a major version so KIP-848 default-protocol behavior in CI does not shift under you when a new Kafka image is published.

Overall the direction is solid — the partition-assignment-strategy drop is the one I would treat as a blocker before merge.

---

Automated review by Claude Code. I may be wrong — please verify against the actual librdkafka behavior for your Kafka version.

[claude]

Review: GH-4190 Add support to KIP-848

Thanks for this — the IGroupProtocol abstraction to split classic vs consumer (KIP-848) group coordination is a clean approach, and moving SessionTimeoutMs/PartitionAssignmentStrategy out of the shared ConsumerConfig is exactly right, since librdkafka rejects those keys when group.protocol=consumer. Migrating CI/compose to KRaft + apache/kafka is also necessary for KIP-848. A few observations below.

🐛 Correctness / behaviour

1. Silent precedence between KafkaSubscription properties and GroupProtocol. When a caller supplies a GroupProtocol, the subscription own SessionTimeout and PartitionAssignmentStrategy (and the sessionTimeout/partitionAssignmentStrategy ctor params) are silently ignored — they only feed the fallback ClassicGroupProtocol built when groupProtocol == null (KafkaMessageConsumer.cs:203-207). So someone who sets both SessionTimeout = 30s and GroupProtocol = new ClassicGroupProtocol() unexpectedly gets Kafka default session timeout, not 30s. Consider either documenting this precedence on both properties, or having the classic fallback merge the subscription values when the supplied ClassicGroupProtocol leaves them unset.

2. ConsumerGroupProtocol.Apply overwrites unconditionally. It always assigns config.GroupRemoteAssignor and config.GroupInstanceId even when the properties are null (ConsumerGroupProtocol.cs:47-48). Harmless today because Apply runs before configHook, but ClassicGroupProtocol guards with HasValue for symmetry — worth being consistent so a future reordering does not clobber a hook-set value.

🧪 Test coverage (per CLAUDE.md TDD guidance)

The only coverage for the new types is the integration KafkaConsumerMessageGatewayProvider, which requires a live KIP-848-capable broker and only ever exercises new ConsumerGroupProtocol() with all-null properties. There are no fast unit tests asserting the Apply behaviour, and this is pure, broker-free logic ideal for TDD:

• ClassicGroupProtocol.Apply sets GroupProtocol.Classic, applies PartitionAssignmentStrategy, and sets SessionTimeoutMs/HeartbeatIntervalMs only when the TimeSpan? has a value.
• ConsumerGroupProtocol.Apply sets GroupProtocol.Consumer and — critically — does not set PartitionAssignmentStrategy or SessionTimeoutMs (the whole reason this PR exists). A regression test locking that in would protect the KIP-848 contract.
• GroupRemoteAssignor / GroupInstanceId / HeartbeatInterval are never exercised anywhere.

Given CLAUDE.md marks the TDD workflow as mandatory, a small unit-test class over Apply() would meaningfully raise confidence here.

🎨 Style / minor

3. Misleading property name. ClassicGroupProtocol.SessionTimeoutMs is a TimeSpan?, but the Ms suffix reads as milliseconds-as-int — and the sibling HeartbeatInterval (also a TimeSpan?) has no such suffix. Renaming to SessionTimeout would be consistent and less surprising. This is brand-new public API, so now is the moment.

4. Unpinned broker image. Both ci.yml and docker-compose-kafka.yaml now use apache/kafka:latest. KIP-848 defaults/behaviour are version-sensitive, so :latest makes CI non-reproducible and can break with no code change. Recommend pinning to a specific tag (e.g. apache/kafka:4.0.0). Not a regression (the prior config also used :latest), just a good moment to pin.

5. docker-compose-kafka.yaml has a trailing space after apache/kafka:latest and is missing a trailing newline at EOF.

✅ Nice touches

• KafkaMessageAssertion.NormalizeToUtcSecond is a more robust timestamp comparison than the previous ToString("yyyy-MM-ddTHH:mm:ss").
• Splitting the generated tests into Classic and Consumer provider trees so both protocols run through the same suite is a good way to get real coverage of the new path.

Overall the design is sound and the core config change is correct — the main asks are unit tests for the Apply logic and clarifying/merging the subscription-vs-GroupProtocol precedence.

🤖 Generated with Claude Code

[claude]

Code Review — GH-4190 Add support to KIP-848

Thanks for this! The IGroupProtocol abstraction is a clean way to introduce the new broker-driven consumer group protocol (KIP-848) alongside the legacy classic protocol, and the backward-compatibility handling (obsolete attributes + null back-filling) is thoughtful. Confluent.Kafka 2.14.0 supports GroupProtocol.Consumer, so the version requirement is satisfied. Below is my feedback.

👍 Strengths

• Clean, small abstraction (IGroupProtocol + ClassicGroupProtocol + ConsumerGroupProtocol), each implementation only sets what its protocol supports (the consumer protocol correctly omits session.timeout.ms / partition.assignment.strategy, which are broker-managed under KIP-848).
• Backward compatible: existing constructor params are preserved and marked [Obsolete] with a clear migration message; classic path behavior is retained by back-filling SessionTimeout/PartitionAssignmentStrategy from the old params.
• Good test coverage — the generated suites are duplicated across Classic and Consumer variants (Proactor + Reactor), and test-configuration.json wires both, so both protocols get exercised.
• The KRaft migration in docker-compose-kafka.yaml/ci.yml is necessary (KIP-848 requires a modern KRaft broker), and dropping ZooKeeper is the right call.

🔧 Issues & suggestions

1. (Medium) KafkaMessageConsumer mutates the caller-supplied IGroupProtocol instance

if (groupProtocol is ClassicGroupProtocol classicGroupProtocol)
{
    classicGroupProtocol.SessionTimeout ??= sessionTimeout;
    classicGroupProtocol.PartitionAssignmentStrategy ??= partitionAssignmentStrategy;
}

This back-fills onto the user's object. Since KafkaSubscription.GroupProtocol is a single shared instance and KafkaMessageConsumerFactory.Create is called once per performer (noOfPerformers), the same instance is mutated by the first consumer created. It's functionally benign today (all performers of one subscription share the same sessionTimeout), but mutating a caller-owned config object as a side effect of constructing a consumer is surprising and could bite if a ClassicGroupProtocol is reused across subscriptions with different session timeouts (first-writer-wins). Consider applying the fallback to a local copy rather than mutating the passed-in instance.

2. (Minor) Silent drop of deprecated params when a ConsumerGroupProtocol is supplied
When a user passes GroupProtocol = new ConsumerGroupProtocol(), the constructor-level sessionTimeout and partitionAssignmentStrategy (which still default to 10s / RoundRobin) are silently ignored, because only ClassicGroupProtocol consumes them. This is correct for KIP-848, but worth calling out explicitly in the XML docs so users migrating don't expect those knobs to take effect.

3. (Minor) No groupProtocol constructor parameter on KafkaSubscription
GroupProtocol is settable only via object initializer, while every other setting has a constructor parameter. Fine if intentional (the constructor is already large), but it's an ergonomic inconsistency — the generated test providers have to use the initializer form ({ GroupProtocol = new ConsumerGroupProtocol() }).

4. (Minor) License-header style inconsistency in the new files
IGroupProtocol.cs, ClassicGroupProtocol.cs, and ConsumerGroupProtocol.cs use // line-comment headers, whereas the rest of the Kafka gateway uses the #region Licence /* ... */ block form. Worth aligning for consistency.

5. (Minor) CI vs docker-compose divergence for schema-registry
docker-compose-kafka.yaml updates SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS to PLAINTEXT://kafka:29092, but .github/workflows/ci.yml still uses kafka:29092. Also KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL is a Confluent-specific var that apache/kafka will ignore (harmless). Keeping the two files in sync avoids confusion.

6. (Nit) Unpinned apache/kafka:latest
Using :latest risks non-reproducible CI when a new broker image ships (KIP-848 client/broker compatibility is exactly the kind of thing that can shift). This matches the pre-existing convention, so non-blocking, but pinning a major version would harden CI.

7. (Nit / pre-existing, not introduced here) In KafkaMessageConsumer, maxPollInterval ??= TimeSpan.FromSeconds(10) and its XML doc ("Defaults to 10s") disagree with KafkaSubscription.MaxPollInterval (defaults to 300000ms). Not part of this change, but since you're in the file it might be worth a follow-up.

Summary

The core design is sound and the feature is well-tested. The main thing I'd address before merge is #1 (mutation of the caller's protocol object); the rest are minor polish. Nice work adding KIP-848 support.

🤖 Automated review — please verify suggestions against your own judgment.

[claude]

Review: GH-4190 Add support to KIP-848

Thanks for this — introducing an IGroupProtocol strategy to split the classic group protocol from the new KIP-848 consumer protocol is a clean, well-scoped design, and the backward-compatibility story (default ClassicGroupProtocol back-filled from the existing constructor params) is handled thoughtfully. The obsoletion of SessionTimeout/PartitionAssignmentStrategy on KafkaSubscription with #pragma warning disable CS0618 at the (few) internal call sites is the right way to deprecate without breaking callers. Below are observations, ordered roughly by importance.

🟠 Correctness / design

1. The caller-supplied GroupProtocol instance is mutated in-place (KafkaMessageConsumer ctor, ~lines 203-208).

groupProtocol ??= new ClassicGroupProtocol();
if (groupProtocol is ClassicGroupProtocol classicGroupProtocol)
{
    classicGroupProtocol.SessionTimeout ??= sessionTimeout;
    classicGroupProtocol.PartitionAssignmentStrategy ??= partitionAssignmentStrategy;
}

KafkaSubscription.GroupProtocol is a single shared instance. When a subscription has more than one performer, KafkaMessageConsumerFactory.Create is called once per consumer and hands the same object to each one, so the constructor back-fills (and later Apply-reads) a shared, caller-owned object. In practice the values are identical so there is no functional bug, but:

• It is a surprising side effect on an object the caller owns (e.g. reusing one ClassicGroupProtocol across two subscriptions lets the first consumer params "win" via ??=).
• If performers are constructed concurrently there is a benign-but-real data race on those ??= writes.

Consider back-filling onto a local copy, or documenting explicitly that a GroupProtocol instance must not be shared. The XML doc on ClassicGroupProtocol describes the back-fill (good), but does not call out the shared-instance implication.

2. ConsumerGroupProtocol + KIP-848 config hygiene — looks correct, worth a confirming note. Nicely, the base _consumerConfig no longer sets SessionTimeoutMs or PartitionAssignmentStrategy, so when ConsumerGroupProtocol is selected none of the classic-only settings that KIP-848 rejects are present. MaxPollIntervalMs (still set) is permitted under the consumer protocol, so that is fine. One thing to note: when a user picks ConsumerGroupProtocol but also set the (now-deprecated) partitionAssignmentStrategy on the subscription, that value is silently ignored. That is the correct outcome for KIP-848, just silent — a brief remark in the docs would help.

🟡 CI / infrastructure

3. KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL on the apache/kafka image (ci.yml + docker-compose). The migration from cp-enterprise-kafka (ZooKeeper) to apache/kafka (KRaft) is the right move for KIP-848 testing. But KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL is a Confluent-platform variable; the Apache image maps KAFKA_* vars to broker properties and will treat confluent.schema.registry.url as an unknown config (logged warning, ignored). Harmless but dead config — probably safe to drop.

4. image: apache/kafka:latest is unpinned. Using :latest in CI means a broker upgrade can silently change test behavior (the pre-existing images were also :latest, so this is not a regression). Pinning to a concrete version (e.g. apache/kafka:3.9.0) would make CI reproducible.

5. Minor: docker-compose-kafka.yaml lost its trailing newline.

🟢 Nits

• ClassicGroupProtocol.Apply assigns config.PartitionAssignmentStrategy = PartitionAssignmentStrategy; unconditionally (even when null). Harmless (null → broker default) and consistent overall, but slightly asymmetric with the HasValue guards used for SessionTimeout/HeartbeatInterval.

✅ Tests

• Good coverage addition: the Consumer gateway provider + generated Proactor/Reactor suites exercise the new protocol end-to-end, and the Standard → Classic rename keeps the existing path covered.
• RetryableChannelSync now retrying on transient UnknownTopicOrPart is a sensible hardening for KRaft/CI timing, and normalizing timestamps to UTC-second in KafkaMessageAssertion removes a real source of flakiness.
• Gap: the pure configuration logic is not unit-tested. ClassicGroupProtocol.Apply, ConsumerGroupProtocol.Apply, and the constructor back-fill (??=) precedence can all be asserted against a ConsumerConfig without a broker, and would lock in exactly the KIP-848 invariant in (2) — e.g. "with ConsumerGroupProtocol, SessionTimeoutMs/PartitionAssignmentStrategy are never set." Worth adding since the broker-backed tests only run when Kafka is available.

Overall this is a solid, well-documented feature. The shared-instance mutation (1) is the item I would most like to see addressed or explicitly documented; the rest are minor.

Reviewed with Claude Code.

[claude]

Review: GH-4190 Add support to KIP-848

Thanks for this — the final shape (an IGroupProtocol strategy with ClassicGroupProtocol / ConsumerGroupProtocol) is a much nicer public API than the enum-plus-three-loose-params approach the first commit started with. A few notes below, mostly minor. Overall this looks solid.

👍 What's good

• Clean abstraction. Moving protocol wiring behind IGroupProtocol.Apply(ConsumerConfig) keeps KafkaMessageConsumer from growing more if (protocol == …) branches, and lets users drop in a fully-configured protocol object.
• Backward compatibility handled deliberately. SessionTimeout and PartitionAssignmentStrategy on KafkaSubscription are [Obsolete] with doc pointers to their replacements, and the constructor back-fills a default ClassicGroupProtocol from the legacy params, so existing callers keep working.
• KIP-848 contract respected. The consumer protocol path deliberately does not set partition.assignment.strategy or session.timeout.ms (which librdkafka rejects under group.protocol=consumer), while the classic path still does. Confluent.Kafka 2.14.0 supports the consumer protocol, so this is wired correctly.
• MIT headers and XML docs are present and detailed.

🔍 Suggestions / questions

1. The consumer mutates a caller-owned protocol object (medium).
In KafkaMessageConsumer:
```csharp
groupProtocol ??= new ClassicGroupProtocol();
if (groupProtocol is ClassicGroupProtocol classicGroupProtocol)
{
classicGroupProtocol.SessionTimeout ??= sessionTimeout;
classicGroupProtocol.PartitionAssignmentStrategy ??= partitionAssignmentStrategy;
}
```
When the protocol comes from KafkaSubscription.GroupProtocol, this mutates the caller's instance. With noOfPerformers > 1, several consumers are created from the same subscription and therefore share the same IGroupProtocol object. The ??= makes it idempotent so it's not a functional bug today, but mutating user-supplied config is a surprising side effect (e.g. the object then reports a SessionTimeout the user never set). Consider back-filling into a local copy, or reading the fallbacks without writing them back.

2. Test coverage is broker-only for the new logic.
The Apply methods and the constructor back-fill are the heart of this change, yet the only coverage is the generated integration tests, which need a running Kafka broker. A couple of cheap, broker-free unit tests would lock down the KIP-848 contract and guard against regressions:

• ClassicGroupProtocol.Apply sets GroupProtocol.Classic, PartitionAssignmentStrategy, SessionTimeoutMs (and HeartbeatIntervalMs when set).
• ConsumerGroupProtocol.Apply sets GroupProtocol.Consumer, GroupRemoteAssignor, GroupInstanceId, and crucially leaves PartitionAssignmentStrategy/SessionTimeoutMs unset.
• The consumer back-fills a default ClassicGroupProtocol when groupProtocol is null.

Note ClassicGroupProtocol.HeartbeatInterval is a new public property with no exercise at all.

3. Minor consistency nit in ConsumerGroupProtocol.Apply.
It assigns GroupRemoteAssignor and GroupInstanceId unconditionally (even when null), whereas ClassicGroupProtocol guards with HasValue. Harmless, but worth aligning the two for readability.

4. Silent drop of deprecated params under the consumer protocol.
If a user selects ConsumerGroupProtocol but also sets the (deprecated) sessionTimeout / partitionAssignmentStrategy, those are silently ignored — correct for KIP-848, but not called out. A one-line remark in the ConsumerGroupProtocol or GroupProtocol property docs would save someone a debugging session.

5. Broker version for the new Consumer tests.
docker-compose-kafka.yaml uses apache/kafka:latest. The consumer group protocol is GA/default-on in Kafka 4.0, so latest should work, but it isn't pinned or explicitly gated (no group.coordinator.rebalance.protocols). If CI ever pulls a broker older than 4.0, the new Consumer/ generated tests would fail confusingly. Worth pinning the image or adding the enablement env var to be safe.

Nice work overall — the API evolution and deprecation handling are the strong points here.

[claude]

Code Review — GH-4190 Add support for KIP-848

Thanks for this! The IGroupProtocol strategy is a clean way to model classic vs. the new (KIP-848) consumer group protocol, and I like that the migration path is handled with [Obsolete] markers on the old KafkaSubscription properties rather than a hard break. XML docs are thorough. A few things worth considering below.

Correctness / design

1. The consumer constructor mutates the caller-supplied IGroupProtocol.
   In KafkaMessageConsumer:

   if (groupProtocol is ClassicGroupProtocol classicGroupProtocol)
   {
       classicGroupProtocol.SessionTimeout ??= sessionTimeout;
       classicGroupProtocol.PartitionAssignmentStrategy ??= partitionAssignmentStrategy;
   }

   groupProtocol is kafkaSubscription.GroupProtocol — a single instance that lives on the (typically long-lived, shared) subscription. A subscription with more than one performer creates multiple consumers from the same subscription, so this back-fill mutates shared state. It's idempotent for a single subscription (all consumers pass the same sessionTimeout), so it's harmless in the common case, but sharing one ClassicGroupProtocol instance across subscriptions with different session timeouts would silently let the first one win. Consider cloning the protocol before back-filling, or documenting that GroupProtocol instances must not be shared.

2. ClassicGroupProtocol.Apply assigns PartitionAssignmentStrategy unconditionally but guards SessionTimeout/HeartbeatInterval.

   config.PartitionAssignmentStrategy = PartitionAssignmentStrategy; // set even when null
   if (SessionTimeout.HasValue) { ... }

   Assigning null to the nullable ConsumerConfig property is a no-op in Confluent.Kafka, so this isn't a bug — but the asymmetry reads oddly. Worth making the three properties consistent for clarity.

3. No client-side guard/docs for the consumer protocol's broker requirement. KIP-848's consumer group protocol requires a sufficiently recent broker (Kafka 4.0, or 3.7+ early access) and librdkafka. If someone points ConsumerGroupProtocol at an older broker they'll get a cryptic librdkafka error at runtime. A note on the minimum supported Kafka version in the ConsumerGroupProtocol XML docs would save users a debugging session.

Test coverage

4. No fast unit tests for the new Apply/back-fill logic. The only exercise of ClassicGroupProtocol.Apply, ConsumerGroupProtocol.Apply, and the ??= back-fill is via the broker-dependent generated integration tests. Given the repo's TDD emphasis (CLAUDE.md), a couple of plain unit tests asserting that Apply sets the expected ConsumerConfig fields (GroupProtocol, GroupRemoteAssignor, GroupInstanceId, SessionTimeoutMs, PartitionAssignmentStrategy) — and that back-fill leaves explicitly-set values untouched — would be fast, deterministic, and would document the contract.

5. Infrastructure-missing tests were removed rather than fixed. The When_infrastructure_missing_and_{validate,assume}_channel_should_throw_exception tests are deleted for Standard/PartitionKey and HasSupportToValidateInfrastructure: false is set across the board. The commit messages explain the reason (Kafka completes async for non-existent topics, and RetryableChannel suppresses the exception). That's a legitimate reason to disable them, but it does drop coverage of genuine validate/assume behavior. Consider tracking a follow-up issue so this doesn't silently stay off.

CI / infra

6. :latest image tags reduce reproducibility. The move to KRaft (apache/kafka:latest) is correct for KIP-848, but pinning to an explicit tag (e.g. apache/kafka:3.9.0) for both the CI service and docker-compose-kafka.yaml avoids surprise breakage when a new Kafka release changes defaults. (The pre-existing Confluent images use :latest too — same concern.)

7. Leftover Confluent-specific env vars on the apache/kafka image. KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL and KAFKA_DELETE_TOPIC_ENABLE are Confluent-image conventions; apache/kafka won't interpret KAFKA_CONFLUENT_*, and delete.topic.enable is effectively always-on under KRaft. They're harmless but dead config now — worth pruning to avoid confusion.

Nits

8. KIP number mismatch in history. The PR title says KIP-848 (correct — the next-gen consumer rebalance protocol), but the branch name and first commit say "KIP-484" (an unrelated KIP). Cosmetic, but confusing for future archaeology.
9. docker-compose-kafka.yaml and test-configuration.json lost their trailing newlines.
10. ClassicGroupProtocol.HeartbeatInterval is only reachable by constructing the protocol manually (no subscription-level or constructor exposure). Fine as-is, just noting the asymmetry with SessionTimeout.

Nothing here is blocking on the feature logic itself — the core IGroupProtocol design is sound. The main asks are the shared-mutation caveat (1) and adding fast unit tests for the new mapping logic (4).

🤖 Generated with Claude Code

[claude]

Code Review — GH-4190: Add support for KIP-848 (Consumer Group Protocol)

Thanks for this — the introduction of IGroupProtocol with ClassicGroupProtocol / ConsumerGroupProtocol is a clean strategy pattern, the backward-compatibility handling (obsolete attributes + ??= back-fill) is thoughtful, and the XML docs are excellent. A few observations below, mostly minor.

👍 Strengths

• Good separation of concerns. IGroupProtocol.Apply(ConsumerConfig) keeps protocol-specific knobs out of KafkaMessageConsumer, and importantly ConsumerGroupProtocol.Apply deliberately avoids setting SessionTimeoutMs/PartitionAssignmentStrategy — those are broker-managed under KIP-848 and would otherwise be rejected by the client. Nice guard.
• Non-breaking migration. Marking SessionTimeout/PartitionAssignmentStrategy [Obsolete] while still honouring them via the back-fill (and wrapping the reads in #pragma warning disable CS0618) preserves existing behaviour.
• KafkaMessageAssertion UTC normalization is a genuine correctness improvement over the previous ToString("yyyy-MM-ddTHH:mm:ss") on a non-normalized DateTimeOffset.

🐛 Correctness / design notes

1. Shared-mutable-state on the subscription's GroupProtocol instance. In KafkaMessageConsumer the back-fill mutates the passed-in instance:

   classicGroupProtocol.SessionTimeout ??= sessionTimeout;
   classicGroupProtocol.PartitionAssignmentStrategy ??= partitionAssignmentStrategy;

   Because KafkaMessageConsumerFactory.Create passes kafkaSubscription.GroupProtocol (a single shared object on the subscription), the first consumer created "bakes in" these values on the shared instance. If a subscription were ever reused to build consumers with different sessionTimeout values, later consumers would silently inherit the first one. Low risk in practice (values come from the same subscription), but consider cloning the protocol before mutating, or documenting that GroupProtocol instances must not be shared.

2. ClassicGroupProtocol.Apply sets PartitionAssignmentStrategy unconditionally, so if someone constructs a ClassicGroupProtocol directly and passes it to the consumer constructor without the subscription back-fill path, config.PartitionAssignmentStrategy could be set to null. That happens to be a valid "use broker default" for the client, but it's worth confirming that's intended vs. mirroring the HasValue pattern used for the timeouts.

🧪 Test coverage

• There are no unit tests for the new ClassicGroupProtocol.Apply, ConsumerGroupProtocol.Apply, or the back-fill/??= logic in the consumer constructor. The only coverage is the generated integration tests via KafkaConsumerMessageGatewayProvider, which require a live broker and won't run in most local/PR contexts. CLAUDE.md mandates a test-first workflow — a couple of fast, broker-free unit tests asserting the resulting ConsumerConfig (e.g. "consumer protocol does not set SessionTimeoutMs", "classic protocol back-fills session timeout from constructor param") would lock in exactly the guarantees this PR relies on.
• The When_infrastructure_missing_and_{validate,assume}_channel_should_throw_exception tests are intentionally absent for the Consumer provider (flagged via HasSupportToValidateInfrastructure: false), per the commit notes about KIP-848 completing asynchronously for non-existent topics. Reasonable, but it does leave the new path with less missing-infra coverage — worth a follow-up to confirm the intended behaviour is captured somewhere.

⚙️ CI / infrastructure

• Unpinned latest tags in both ci.yml and docker-compose-kafka.yaml (apache/kafka:latest, confluentinc/cp-schema-registry:latest). latest makes CI non-reproducible and can break unexpectedly when upstream publishes a new image. Recommend pinning to explicit versions.
• KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL is a Confluent-image-specific variable and is likely ignored by the apache/kafka image — harmless, but noise. The # 1. / # 2. / # 3. comment blocks also read like they were pasted from a tutorial and could be trimmed.
• docker-compose-kafka.yaml is missing a trailing newline at EOF.

📝 Nits

• The PR title references KIP-848 (correct — the Next-Gen Consumer Rebalance Protocol), but the commit messages and branch name say "KIP-484", which is an unrelated KIP. Just a typo, but worth correcting for searchability.
• ConsumerGroupProtocol.Apply assigns GroupRemoteAssignor/GroupInstanceId even when null; harmless (null = broker default) but slightly noisy.

Overall this is a solid, well-documented feature. The main asks are (1) add unit-level tests for the protocol Apply/back-fill logic per the repo's TDD guidance, and (2) pin the Docker image tags. 👍

Reviewed by Claude — automated PR review.

[claude]

Review of #4233 — Add support for KIP-848 (Kafka new consumer group protocol)

Thanks for this — the IGroupProtocol abstraction is a clean way to introduce the KIP-848 consumer protocol without breaking the classic path, and the correct handling of the mutually-exclusive config keys (not applying session.timeout.ms / heartbeat.interval.ms / partition.assignment.strategy when the consumer protocol is selected) is the crux of the feature and it's done right. The obsoleting of SessionTimeout / PartitionAssignmentStrategy on KafkaSubscription with #pragma guards at the call sites is tidy. A few things worth considering below.

Correctness / design

1. Shared IGroupProtocol instance is mutated as a side effect. In KafkaMessageConsumer the passed protocol is back-filled in place:

   groupProtocol ??= new ClassicGroupProtocol();
   if (groupProtocol is ClassicGroupProtocol classicGroupProtocol)
   {
       classicGroupProtocol.SessionTimeout ??= sessionTimeout;
       classicGroupProtocol.PartitionAssignmentStrategy ??= partitionAssignmentStrategy;
   }

   KafkaSubscription.GroupProtocol holds a single instance, and Brighter creates one consumer per performer thread from the same subscription — so the same object is mutated by every consumer. For ClassicGroupProtocol the back-fill is idempotent, so it's harmless in practice, but it's a hidden mutation of the caller's object. If someone reuses one ClassicGroupProtocol across two subscriptions with different sessionTimeout values, the first back-fill wins and the second silently inherits it (??=). Consider back-filling onto a local copy, or documenting that a GroupProtocol instance should not be shared across subscriptions.

2. Static membership + shared instance is a footgun. ConsumerGroupProtocol.GroupInstanceId maps to Kafka static membership, which requires a unique group.instance.id per consumer instance. Because the subscription's single GroupProtocol (with a fixed GroupInstanceId) is applied to every consumer created from that subscription, all performers would register the same static id and collide on group join. This is inherent to placing GroupInstanceId on a shared subscription object — worth at least an XML-doc warning that GroupInstanceId should only be set when there is a single consumer per subscription.

3. ClassicGroupProtocol.Apply sets PartitionAssignmentStrategy unconditionally (even when null), whereas the two timeouts use HasValue guards. It's harmless because the property is nullable, but the asymmetry is slightly surprising — consider guarding it the same way for consistency.

Test coverage

4. No unit test for the Apply mapping. The new Consumer suite is all integration tests that need a running KRaft broker. A cheap, infra-free unit test over ClassicGroupProtocol.Apply / ConsumerGroupProtocol.Apply would lock in the KIP-848 contract that matters most — i.e. that the consumer protocol does not set SessionTimeoutMs / HeartbeatIntervalMs / PartitionAssignmentStrategy, since librdkafka errors at consumer creation if those are present alongside group.protocol=consumer. Given the repo's TDD emphasis (CLAUDE.md), this would be a valuable guard against regressions.

5. Dropped infra-validation tests for the Consumer path. The HasSupportToValidateInfrastructure: false flag (and the add-then-revert churn in the commit history) reflect that the consumer protocol completes asynchronously for non-existent topics, so the "missing infrastructure throws" behaviour doesn't hold there. That's a reasonable call, but it does mean the KIP-848 path has weaker infra-validation guarantees than classic — good to call out explicitly for maintainers/users.

CI / infrastructure

6. :latest image tags. The move to apache/kafka:latest (KRaft, no ZooKeeper) is the right direction, but the broker, schema-registry, and control-center all use :latest. Non-pinned tags make CI non-reproducible and can break unrelated PRs when upstream publishes a new image. Consider pinning to explicit versions.

7. Cosmetics in docker-compose-kafka.yaml: trailing whitespace after apache/kafka:latest , and the file lost its trailing newline. Minor.

8. ClassicGroupProtocol.Apply has a double space: config.HeartbeatIntervalMs = Convert.ToInt32(...). Trivial.

Nice touches

• KafkaMessageAssertion.NormalizeToUtcSecond is a good fix — comparing normalized-to-UTC seconds is more robust than the previous ToString("yyyy-MM-ddTHH:mm:ss") comparison, which was timezone-sensitive.
• XML documentation on the new types is thorough and explains the back-fill precedence clearly.

Overall this is a solid, well-scoped feature. The main things I'd like to see addressed are the shared-instance mutation semantics (items 1–2) and a small unit test over Apply (item 4); the rest are minor.

_{🤖 Automated review}

[claude]

Code Review — KIP-848 support (Kafka consumer group protocol)

Thanks for this — it's a clean, well-considered feature. The IGroupProtocol abstraction is a good fit, backward compatibility is preserved via [Obsolete] + a default ClassicGroupProtocol, and the XML docs are unusually thorough (the footgun callouts on shared instances and static membership are excellent). A few observations below, mostly non-blocking.

👍 Strengths

• IGroupProtocol cleanly separates classic vs. consumer protocol config; Apply(ConsumerConfig) is easy to reason about and unit test.
• Deprecation of KafkaSubscription.SessionTimeout / PartitionAssignmentStrategy with #pragma warning disable CS0618 at the internal call sites is done correctly — no unsuppressed obsolete usages leak into the build.
• ConsumerGroupProtocol correctly avoids emitting session.timeout.ms / heartbeat.interval.ms / partition.assignment.strategy (librdkafka rejects those alongside group.protocol=consumer), and Should_not_set_classic_only_settings locks that in.
• Nice touch disabling HasSupportToValidateInfrastructure for the Consumer generator config — consistent with the documented weaker validation guarantees of broker-driven assignment.

🔎 Suggestions / questions

1. In-place mutation of the caller's ClassicGroupProtocol (design, medium).
KafkaMessageConsumer back-fills the supplied instance in place (SessionTimeout ??= …, PartitionAssignmentStrategy ??= …). You've documented the "don't share an instance across subscriptions" hazard well, but the safer fix is to not require the docs at all: shallow-copy the protocol inside the consumer before back-filling. The subscription holds one instance and the factory hands it to every performer's consumer, so the current contract puts the burden on the user to avoid a subtle cross-subscription bleed. A defensive clone would eliminate the footgun entirely.

2. Broker enablement of the consumer protocol (CI, please verify).
The new apache/kafka:4.0.2 broker config (both ci.yml and docker-compose-kafka.yaml) doesn't explicitly set group.coordinator.rebalance.protocols to include consumer. If that isn't on by default in 4.0.2, the Consumer generated tests would fail when librdkafka sends group.protocol=consumer. Can you confirm the Consumer suite actually goes green in CI against this image (i.e. the protocol is really being exercised, not silently falling back)?

3. Mixed-vendor container stack (infra, minor).
The broker is now apache/kafka while schema-registry (8.0.6) and control-center (7.9.8) remain Confluent images. Also KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL is a Confluent-image env var that the apache/kafka image ignores — looks like dead config that can be dropped.

4. Missing unit coverage for the actual behavior change (test coverage).
The Apply() mappings are unit-tested in isolation, but the new integration logic in the KafkaMessageConsumer constructor — (a) null groupProtocol ⇒ ClassicGroupProtocol back-filled from sessionTimeout/partitionAssignmentStrategy, and (b) a supplied ConsumerGroupProtocol leaves the classic-only settings unset — is only covered by broker-dependent integration tests. Since configHook runs right after Apply (line 213), you can capture the resulting ConsumerConfig in a fast unit test:

ConsumerConfig? captured = null;
_ = new KafkaMessageConsumer(config, routingKey, groupId: "g",
        groupProtocol: new ConsumerGroupProtocol(),
        configHook: c => captured = c);
Assert.Equal(GroupProtocol.Consumer, captured!.GroupProtocol);
Assert.Null(captured.SessionTimeoutMs);            // classic-only setting not leaked
Assert.Null(captured.PartitionAssignmentStrategy);

This would guard the regression that matters most cheaply and without a broker.

5. Minor consistency nit.
ConsumerGroupProtocol.Apply assigns GroupRemoteAssignor / GroupInstanceId unconditionally (harmless null no-op), while ClassicGroupProtocol.Apply guards with HasValue. Not a bug, just two different idioms for the same "don't write when unset" intent.

Overall: solid work, backward-compatible, and well-documented. My only near-blocking item is confirming the CI broker actually enables the consumer protocol (#2); the clone (#1) and the constructor-level unit test (#4) would be worthwhile hardening.

_{🤖 Automated review — generated with Claude Code}