MINOR: Document that REMAIN_IN_GROUP is ignored with streams protocol #21433
Conversation
When using the streams group protocol (group.protocol=streams), the consumer always leaves the group on close, regardless of whether REMAIN_IN_GROUP was specified. This commit documents this behavior in the javadocs for CloseOptions, KafkaStreams.close(), and KafkaStreams.close(Duration), and adds an INFO log message when this occurs.
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Overall LGTM. Just a nit question. Feel free to merge either way.
| try { | ||
| final GroupMembershipOperation membershipOperation = | ||
| leaveGroupRequested.get() == org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP ? LEAVE_GROUP : REMAIN_IN_GROUP; | ||
| if (membershipOperation == REMAIN_IN_GROUP && streamsRebalanceData.isPresent()) { |
There was a problem hiding this comment.
Not 100% sure about re-using streamsRebalanceData.isPresent() -- Sure it works, but is it "clean"?
Should we rather introduce a member-variable (eg GroupProtocol rebalanceProtocol)?
There was a problem hiding this comment.
The pattern streamsRebalanceData.isPresent() is already used throughout this file to check for the streams protocol (lines 1122, 1128, 1178, 1332, 1503). Introducing a separate GroupProtocol member variable would add redundancy since streamsRebalanceData is only populated when using the streams protocol.
If you think we should change the pattern, it would be a larger cleanup across the file. Happy to do that as a separate PR if you prefer, but for this minor doc change I'd prefer to stay consistent with the existing pattern.
2 participants
Summary
When using the streams group protocol (
group.protocol=streams), theconsumer always leaves the group on close, regardless of whether
REMAIN_IN_GROUPwas specified. This differs from the classic protocolbehavior where the consumer does not leave the group explicitly by
default.
This PR documents this behavior in the javadocs for
CloseOptions.GroupMembershipOperation,KafkaStreams.close(), andKafkaStreams.close(Duration), and adds an INFO log message when aclose occurs with the streams protocol to inform users about this
behavior.
Test plan
Reviewers: Matthias J. Sax matthias@confluent.io