lucasbru commented on code in PR #22676:
URL: https://github.com/apache/kafka/pull/22676#discussion_r3497728293
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java:
##########
@@ -1808,6 +1816,10 @@ public void testCleanupGroupMetadataForShareGroup() {
when(groupMetadataManager.groupIds()).thenReturn(Set.of("group-id"));
when(groupMetadataManager.group("group-id")).thenReturn(group);
+ // Offset cleanup runs uniformly; for share groups it has nothing to
do and returns
Review Comment:
nit: The real cleanupExpiredOffsets actually returns true for a share group,
not false. I think we should test the real scenario here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java:
##########
@@ -1736,29 +1737,37 @@ public void
testCleanupGroupMetadataTombstoneStreamsGroupWithoutStoredTopologyEp
mock(CoordinatorMetricsShard.class)
);
- StreamsGroup streamsGroup = mock(StreamsGroup.class);
- when(streamsGroup.shouldExpire()).thenReturn(true);
- when(streamsGroup.type()).thenReturn(Group.GroupType.STREAMS);
- when(streamsGroup.storedDescriptionTopologyEpoch()).thenReturn(-1);
+ CoordinatorRecord offsetCommitTombstone =
GroupCoordinatorRecordHelpers.newOffsetCommitTombstoneRecord("group-id",
"topic", 0);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<CoordinatorRecord>> recordsCapture =
ArgumentCaptor.forClass(List.class);
+
+ Group group = mock(Group.class);
+ when(group.shouldExpire(config)).thenReturn(false);
when(groupMetadataManager.groupIds()).thenReturn(Set.of("group-id"));
- when(groupMetadataManager.group("group-id")).thenReturn(streamsGroup);
- when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
any())).thenReturn(true);
+ when(groupMetadataManager.group("group-id")).thenReturn(group);
+ when(offsetMetadataManager.cleanupExpiredOffsets(eq("group-id"),
recordsCapture.capture()))
+ .thenAnswer(invocation -> {
+ recordsCapture.getValue().add(offsetCommitTombstone);
+ return true;
+ });
- coordinator.cleanupGroupMetadata();
+ CoordinatorResult<Void, CoordinatorRecord> result =
coordinator.cleanupGroupMetadata();
- verify(groupMetadataManager,
times(1)).maybeDeleteGroup(eq("group-id"), any());
+ // Offset tombstone went through; group tombstone is deferred.
+ assertEquals(List.of(offsetCommitTombstone), result.records());
+ verify(offsetMetadataManager,
times(1)).cleanupExpiredOffsets(eq("group-id"), any());
+ verify(groupMetadataManager, never()).maybeDeleteGroup(eq("group-id"),
any());
}
@Test
- public void
testCleanupGroupMetadataIgnoresStoredTopologyEpochWhenNoPluginConfigured() {
- // Plugin absent on this broker (operator unset / never set): even a
streams group with
- // a non-default storedEpoch must NOT be deferred — no cleanup cycle
is running to
- // clear it, and leaving the gate engaged would prevent natural
expiration forever.
+ public void testCleanupGroupMetadataTombstonesWhenGroupAgreesToExpire() {
+ // Counterpart to the refuse test: when the group's
shouldExpire(config) returns true
Review Comment:
"Counterpart to the refuse test" - not sure what it references. Maybe just
omit?
--
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]
