lucasbru commented on code in PR #23978:
URL: https://github.com/apache/kafka/pull/23978#discussion_r4403170997
##########
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:
Yes, I'd backport it to 4.4, since KIP-1331 is in 4.4 and it cherry-picks
cleanly. I don't think it needs anything on the website. Clients handle both
errors the same way, so it's only more specific on the wire. I'll amend
KIP-1331 to list the error.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupTopologyDescriptionRequestManager.java:
##########
@@ -165,6 +165,14 @@ private void onResponse(final ClientResponse response,
final Throwable exception
streamsRebalanceData.setTopologyPushRequired(false);
break;
+ case STREAMS_INVALID_TOPOLOGY_EPOCH:
Review Comment:
Done, added it to that list.
##########
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:
Done.
##########
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:
I'm not sure the split buys us much here. The push carries a
TopologyDescription, not the heartbeat Topology, so at the same epoch there's
nothing to compare it against. A member with a different topology at the same
epoch already gets STREAMS_INVALID_TOPOLOGY_EPOCH on the heartbeat and fails
before it pushes. For a stale push, returning NONE without storing it would be
misleading, and the client clears the push flag on either result anyway. Also,
since topology updates aren't supported, the group's topology epoch doesn't
move after initialization, so a mismatch here means a buggy client or a
recreated group. I'd keep a single error for now and revisit when topology
updates land. WDYT?
--
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]
