squah-confluent commented on code in PR #22748:
URL: https://github.com/apache/kafka/pull/22748#discussion_r3665465842
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -18992,6 +18993,82 @@ public void
testStreamsGroupHeartbeatRejectsDuplicateTaskOffsets() {
assertEquals(MemberTaskOffsets.EMPTY, group.taskOffsets(memberId));
}
+ @Test
+ public void
testStreamsGroupHeartbeatRefinementStepBumpsEpochWithoutRunningAssignor() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String subtopology1 = "subtopology1";
+ String fooTopicName = "foo";
+ Uuid fooTopicId = Uuid.randomUuid();
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new
Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
+ ));
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 4)
+ .buildCoordinatorMetadataImage();
+ long groupMetadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ));
+
+ // The member is reconciled (STABLE, at the assignment epoch), yet its
current assignment differs from the
+ // target: it owns partitions {0,1,2} while the target assigns
{0,1,2,3}. With the no-op refiner (which
+ // returns the target as-is), this stands in for "the refiner produced
an intermediate assignment the member
+ // has not reconciled to yet", so the refinement predicate fires and a
REFINE step is taken. Nothing else
Review Comment:
```
// With the no-op refiner (which
// returns the target as-is), this stands in for "the refiner
produced an intermediate assignment the member
// has not reconciled to yet",
```
I can't make sense of this comment
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4358,6 +4421,14 @@ private UpdateTargetAssignmentResult<Map<String,
TasksTuple>> maybeUpdateStreams
return new UpdateTargetAssignmentResult<>(group.assignmentEpoch(),
updatedMembersAndTargetAssignment.targetAssignment());
}
+ if (refineOnly) {
Review Comment:
@mjsax Do you think you could add the `&& group.groupEpoch() >=
group.assignmentEpoch()` guard?
--
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]
