mjsax commented on code in PR #23978:
URL: https://github.com/apache/kafka/pull/23978#discussion_r4396768365
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2504,8 +2505,12 @@ private static void maybeSetTopologyStaleStatus(final
StreamsGroup group, final
*
* - If the topology of the group is uninitialized, it is initialized
with the topology from the request. A corresponding
* record is added to records.
- * - If the topology of the group is initialized, and the request defines
a topology, they are compared. If they
- * are not empty, an InvalidRequestException is thrown.
+ * - If the topology of the group is initialized, and the request defines
a topology with a lower epoch, the
+ * request topology is ignored and the member is reported as stale.
+ * - If the request defines a different topology with the same epoch as
the group, a
+ * StreamsInvalidTopologyEpochException is thrown.
+ * - If the request defines a different topology with a higher epoch than
the group, an InvalidRequestException
+ * is thrown, since topology updates are not supported.
Review Comment:
```suggestion
* is thrown, since topology updates are not supported yet
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -8810,20 +8820,20 @@ private CoordinatorResult<HeartbeatResponseData,
CoordinatorRecord> classicGroup
* @return The matching {@link StreamsGroupMember}.
* @throws GroupIdNotFoundException if no streams group with this id
exists.
* @throws UnknownMemberIdException if the member is not currently in the
group.
- * @throws InvalidRequestException if {@code pushedEpoch} does not match
the group's
- * current topology epoch.
+ * @throws StreamsInvalidTopologyEpochException if {@code pushedEpoch}
does not match the
+ * group's current topology
epoch.
*/
public StreamsGroupMember validateStreamsGroupTopologyDescriptionUpdate(
String groupId,
String memberId,
int pushedEpoch,
long committedOffset
- ) throws GroupIdNotFoundException, UnknownMemberIdException,
InvalidRequestException {
+ ) throws GroupIdNotFoundException, UnknownMemberIdException,
StreamsInvalidTopologyEpochException {
StreamsGroup group = streamsGroup(groupId, committedOffset);
StreamsGroupMember member = group.getMemberOrThrow(memberId,
committedOffset);
int currentEpoch = group.currentTopologyEpoch();
if (pushedEpoch != currentEpoch) {
Review Comment:
Should we be more fine grained similar to "topology epoch in the heartbeat"
and just ignore `pushedEpoch < currentEpoch` as stale, return
`StreamsInvalidTopologyEpochException` only if topology is different on same
epoch, and return `InvalidRequestException` if `pushedEpoch > currentEpoch` ?
##########
clients/src/main/resources/common/message/StreamsGroupTopologyDescriptionUpdateResponse.json:
##########
@@ -32,6 +32,7 @@
// - UNKNOWN_MEMBER_ID (version 0+)
// - GROUP_ID_NOT_FOUND (version 0+)
// - STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED (version 0+)
+ // - STREAMS_INVALID_TOPOLOGY_EPOCH (version 0+)
Review Comment:
SGTM. -- Can we backport this to 4.4.1 ? Do we need to document it
explicitly on the webpage (I would think no, but just wanted to double check).
--
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]
