chia7712 commented on code in PR #22593:
URL: https://github.com/apache/kafka/pull/22593#discussion_r4152701247
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -7930,6 +7934,58 @@ private CoordinatorResult<Void, CoordinatorRecord>
updateStaticMemberThenRebalan
return maybeCompleteJoinPhase(group);
}
+ /**
+ * Parses the set of topics a classic member is subscribed to under the
given protocol. This is used to
+ * detect a subscription change when a static member rejoins during the
Stable stage with the same
+ * selected protocol but a different subscription (for example a
topics.regex member that discovered new
+ * topics). Without it the coordinator would return the cached assignment
and the newly subscribed topics
+ * would never be assigned to the group.
+ *
+ * <p>Only the subscribed topics are extracted. Other parts of the
embedded subscription, such as the
+ * owned partitions or the user data, do not require a new assignment and
must not trigger a rebalance.
+ *
+ * <p>Empty is returned when the group is not using the consumer protocol,
when no protocol is selected,
+ * or when the embedded consumer protocol cannot be parsed. In those cases
the coordinator falls back to
+ * its protocol-based behaviour: two members whose subscriptions are both
empty compare equal and do not
+ * trigger a rebalance. This mirrors {@link
ClassicGroup#computeSubscribedTopics()}, which also leaves the
+ * coordinator unaware of the subscribed topics when the consumer protocol
cannot be parsed.
+ *
+ * @param group The group.
+ * @param protocols The member's supported protocols.
+ * @param protocolName The protocol whose subscription is extracted, or
null if none is selected.
+ * @return The set of subscribed topics, or empty if it cannot be
determined.
+ */
+ private Optional<Set<String>> parseSubscribedTopics(
Review Comment:
Is this logic from ClassicGroup#computeSubscribedTopics? If so, would you
mind adding a helper method to reduce duplicate code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
