Add ConsumerConfig support for group.protocol=consumer (KIP-848) - #613
Add ConsumerConfig support for group.protocol=consumer (KIP-848)#613tobiajo wants to merge 7 commits into
Conversation
ConsumerConfig gains groupProtocol and groupRemoteAssignor fields for the KIP-848 configs group.protocol and group.remote.assignor. kafka-clients rejects partition.assignment.strategy, session.timeout.ms and heartbeat.interval.ms at construction time when group.protocol is set to consumer (assignment and session timing are server-managed under the new protocol), so these are now only emitted under the classic protocol; group.remote.assignor is the mirror image, only emitted under the consumer protocol and rejected by the client under classic. The new fields break binary compatibility of the case class's apply/copy/unapply/constructor for pre-compiled callers; on Scala 2.13 the field count also crosses the 22-field cap on synthesized unapply (22 -> 24), so tuple-style pattern matching against ConsumerConfig no longer compiles there at all. versionPolicyIntention is set to Compatibility.None accordingly: the next release must be a major version bump, after which the setting should be reset to Compatibility.BinaryCompatible. Also drops a stale mimaBinaryIssueFilters entry for Producer.sendOffsetsToTransaction, unrelated to this change: the method has shipped in every release since v20.2.0, so the filter no longer excludes anything.
retry-backoff/retry.backoff.ms were set to 100 in both .conf fixtures, which is exactly CommonConfig.retryBackoff's default -- so if that key's parsing broke silently, the parsing tests wouldn't catch it. Bumped to 123 and added the matching common override to custom. client-rack/client.rack were never set by either fixture, leaving ConsumerConfig.clientRack's parsing entirely unexercised -- and unlike CommonConfig's fields, which CommonConfigSpec covers independently, it has no other test coverage. Added a fixture value for it too. Both gaps pre-date this branch and are unrelated to the group.protocol=consumer feature; kept as a separate commit so it can be reverted independently.
|
Warning Review limit reachedNext included review available in 19 minutes. View limit detailsLimit details: You’ve used the included review currently available. You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. Review configuration: ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
Included review availability: Your plan provides up to 1 included review per hour; 0 remain after this review. 📝 WalkthroughWalkthrough
ChangesConsumer group protocol support
Estimated code review effort: 3 (Moderate) | ~20 minutes Merge Risk: ⚪ Minimal · up to This change adds consumer group protocol configuration and related validation while documenting the intentional compatibility boundary; no actionable merge-blocking risk remains beyond normal checks and review. Sequence Diagram(s)sequenceDiagram
participant Application
participant ConsumerConfig
participant GroupProtocolFromConf
participant KafkaConsumerConfig
Application->>ConsumerConfig: load group protocol settings
ConsumerConfig->>GroupProtocolFromConf: parse protocol name
GroupProtocolFromConf-->>ConsumerConfig: return protocol value
ConsumerConfig->>ConsumerConfig: generate protocol-specific bindings
ConsumerConfig->>KafkaConsumerConfig: validate generated properties
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Full details: Docstring CoverageExplanation No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check. Docstring coverage is scoped to functions touched by this diff. Analyzed 0 functions across 0 files. (1 skipped: 1 unsupported.) ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
3bb2227 to
87767f4
Compare
| // `group.protocol=consumer` rejects `partition.assignment.strategy`, `session.timeout.ms` and | ||
| // `heartbeat.interval.ms` (assignment and session timing are server-managed under the new | ||
| // protocol); `group.protocol=classic` rejects `group.remote.assignor`. | ||
| val protocolSpecificMap = groupProtocol match { |
There was a problem hiding this comment.
If someone sets group-protocol=consumer together with an explicit session-timeout or partition-assignment-strategy, the values parse fine and then get dropped here, so failure detection timing quietly falls back to broker defaults while the operator thinks their tuning is active. Raw kafka-clients fails fast on that combination. The config parser can tell explicit values from defaults, so it could warn or reject there instead of silently ignoring them.
There was a problem hiding this comment.
Given that the existing configurations (sessionTimeout and heartbeatInterval) are not optional in skafka, there is no possibility to reject. The alternative would be an ADT of configuration for different protocols, moving protocol specific params into it, which would be a larger change.
I chose this as I felt it followed existing pattern as there are other cases too, where redundant config is ignored, not sure if comparable to this though as kafka-client not rejecting on those cases.
Would it be preferable with the following?
case class Classic(sessionTimeout: FiniteDuration, heartbeatInterval: FiniteDuration)
case class Consumer(remoteAssignor: Option[String])
Just reacting on configurations that diverge from default for the existing configurations I think would be meaningless, adding false security.
| sslSupport: SslSupportConfig = SslSupportConfig.Default, | ||
| clientRack: Option[String] = None, | ||
| groupProtocol: GroupProtocol = GroupProtocol.Classic, | ||
| groupRemoteAssignor: Option[String] = None, // only emitted under GroupProtocol.Consumer |
There was a problem hiding this comment.
Setting group-remote-assignor without group-protocol=consumer parses fine and the assignor just gets dropped, which you only notice from runtime behaviour. Since GroupProtocol is new in this PR, the assignor could live inside the Consumer case instead, e.g. case class Consumer(remoteAssignor: Option[String] = None). That makes the invalid combination unrepresentable and saves a case class field.
| // On Scala 2.13 this also crosses the 22-field cap on synthesized unapply (22 -> 24), so | ||
| // tuple-style pattern matching against ConsumerConfig no longer compiles there at all. | ||
| // TODO reset to Compatibility.BinaryCompatible after that release | ||
| ThisBuild / versionPolicyIntention := Compatibility.None |
There was a problem hiding this comment.
This is ThisBuild scoped, so bincompat checking is off for every module and the whole API until the TODO is reverted, and nothing enforces that the next tag is actually a major. An unrelated accidental binary break merged before that release would pass versionPolicyCheck silently. Scoping it to the skafka project, or filtering just the ConsumerConfig break, keeps the window smaller.
There was a problem hiding this comment.
Keeping open so whoever merges this does not miss it.
| } | ||
| } | ||
|
|
||
| private implicit val GroupProtocolFromConf: FromConf[GroupProtocol] = FromConf[GroupProtocol] { (conf, path) => |
There was a problem hiding this comment.
This is the third copy of the same find-by-name-or-throw block (AutoOffsetReset and IsolationLevel above), and the next enum config field would copy it a fourth time. Might be worth a small shared helper, something like enumFromConf[T](values: Set[T], label: String)(name: T => String): FromConf[T].
Compatibility.None was set at ThisBuild scope, disabling bincompat checking for every module until the TODO is reverted -- an unrelated accidental binary break in play-json/metrics/metrics-prometheus-v1 would have passed versionPolicyCheck silently. Only skafka itself breaks compatibility (the new ConsumerConfig fields), so the override now lives in that project's settings and the other modules keep being checked against the last release.
toUpperCase without a locale uses the JVM default, so on a Turkish default locale "classic" uppercases to CLASSİC (dotted İ), which never matches what kafka-clients emits and fails the test spuriously.
GroupProtocolFromConf was the third copy of the find-by-name-or-throw block in ConsumerConfig; the next enum config field would copy it a fourth time. The helper is private[skafka] to keep it out of the published API. The remaining copies (CommonConfig, ProducerConfig, KeystoreType) predate this PR and are left for a follow-up.
It sits directly below the helper and is the same find-by-name-or-throw block. The error message loses its quotes around the value; nothing asserts it.
|
2 participants
Summary
groupProtocol/groupRemoteAssignorfields toConsumerConfigfor the KIP-848 configsgroup.protocolandgroup.remote.assignor; the newGroupProtocolsealed trait is modelled onIsolationLevel.group.protocol=consumerrejectspartition.assignment.strategy,session.timeout.msandheartbeat.interval.ms(assignment and session timing are server-managed under KIP-848), and classic rejectsgroup.remote.assignor— so each is now emitted only under the protocol that accepts it. Tests construct the realorg.apache.kafka.clients.consumer.ConsumerConfigunder both protocols to exercise this validation.apply/copy/unapply/constructor (apply/copystay source-compatible via defaults); on Scala 2.13 the field count also crosses the 22-field cap on synthesizedunapply(22 → 24), so tuple-style pattern matching againstConsumerConfigno longer compiles there at all.versionPolicyIntentionis set toCompatibility.None: the next release must be a major, after which it should be reset toCompatibility.BinaryCompatible. AmimaBinaryIssueFiltersentry was not used instead because those filters match by bare method name, not signature, and would permanently mask future unrelated breaks to the same methods.mimaBinaryIssueFiltersentry forProducer.sendOffsetsToTransaction(unrelated; the method has shipped in every release since v20.2.0).ConsumerConfigSpecparsing fixtures:retry-backoffwas set to exactly its default (100 ms), so a silent parse break of that key would pass, andclient-rackwas never set at all, leavingConsumerConfig.clientRackparsing unexercised. Unrelated to the feature; kept separate so it can be reverted independently.NOTE: the below auto-generated summary is quite bad, ignore it.
Summary by CodeRabbit
New Features
Bug Fixes