KIP-1331: Remove remaining rebase conflict duplicates

StreamsConfig had TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG defined twice (once in the field block, once appended at the end of the static ConfigDef builder) and a redundant second .define() call in the builder. GroupCoordinatorServiceTopologyDescriptionTest had eight test methods duplicated from an older version of the POC that called runStreamsGroupTopologyCleanupCycle (renamed to runOneStreamsTopologyCleanupCycle in trunk) — those stale copies are removed; the correct versions earlier in the file are kept. GroupCoordinatorShardTest had three stale testClearStoredDescriptionTopologyEpochIfMatches* tests (using the old method name) and two stale listStreamsGroupsNeedingTopologyCleanup tests (using wrong isEmpty/groupIds/group/allOffsetsExpired call signatures) — all replaced by the already-correct versions in the earlier part of the file.
2026-11-04 04:44:54 +00:00 · 2026-06-29 12:35:33 +02:00
1 parent 2728e2de76
commit fbf5a8c995
@@ -1428,165 +1428,6 @@ public class GroupCoordinatorServiceTopologyDescriptionTest {
        return groups.stream().filter(g -> g.groupId().equals(groupId)).findFirst().orElseThrow();
    }

-    @Test
-    public void testCleanupCycleNoOpWhenNoPlugin() {
-        // No plugin configured -> the cycle must not even touch the runtime.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        GroupCoordinatorService service = buildService(runtime, Optional.empty(), true);
-
-        service.runStreamsGroupTopologyCleanupCycle();
-
-        verify(runtime, never()).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any());
-        verify(runtime, never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any());
-    }
-
-    @Test
-    public void testCleanupCycleClearsStoredEpochOnPluginSuccess() {
-        // Eligibility scan returns one group at storedEpoch=4; plugin succeeds; the cycle must
-        // schedule the conditional clear-stored write echoing the same epoch back so a
-        // concurrent setTopology that has advanced the field is preserved.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class);
-        when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null));
-        when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()))
-            .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
-        when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"), eq(GROUP_TP), any()))
-            .thenReturn(CompletableFuture.completedFuture(null));
-
-        GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true);
-        service.runStreamsGroupTopologyCleanupCycle();
-
-        verify(plugin, times(1)).deleteTopology("foo");
-        verify(runtime, times(1)).scheduleWriteOperation(eq("clear-stored-topology-epoch"), eq(GROUP_TP), any());
-    }
-
-    @Test
-    public void testCleanupCycleSkipsClearOnPluginFailure() {
-        // Plugin fails -> the cycle must NOT clear stored epoch; the group stays gated on
-        // the next sweep and the next cycle retries the plugin call.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class);
-        when(plugin.deleteTopology("foo"))
-            .thenReturn(CompletableFuture.failedFuture(new RuntimeException("plugin offline")));
-        when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()))
-            .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
-
-        GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true);
-        service.runStreamsGroupTopologyCleanupCycle();
-
-        verify(plugin, times(1)).deleteTopology("foo");
-        verify(runtime, never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any());
-    }
-
-    @Test
-    public void testCleanupCycleEmptyEligibility() {
-        // No groups eligible -> plugin is not called and no clear write is scheduled.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class);
-        when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()))
-            .thenReturn(List.of(CompletableFuture.completedFuture(Map.of())));
-
-        GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true);
-        service.runStreamsGroupTopologyCleanupCycle();
-
-        verify(plugin, never()).deleteTopology(anyString());
-        verify(runtime, never()).scheduleWriteOperation(eq("clear-stored-topology-epoch"), any(), any());
-    }
-
-    @Test
-    public void testCleanupCycleSingleFlightSkipsConcurrentCycle() {
-        // The first cycle's per-partition read is parked on an unresolved future. A second
-        // call must observe cleanupCycleInFlight and skip without scheduling another read.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class);
-        when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()))
-            .thenReturn(List.of(new CompletableFuture<>()));
-
-        GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true);
-        service.runStreamsGroupTopologyCleanupCycle();
-        service.runStreamsGroupTopologyCleanupCycle();
-
-        verify(runtime, times(1)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any());
-    }
-
-    @Test
-    public void testCleanupCycleSingleFlightReleasesFlagAfterCycleCompletes() {
-        // The skip case alone does not prove the flag is ever released: a buggy whenComplete
-        // (e.g., missing the partitionDone allOf join) would leave it set forever and silently
-        // disable every subsequent cycle. Drive a full cycle to completion (read resolves,
-        // plugin delete settles, conditional clear write settles), then issue a second cycle
-        // and verify it observes the released flag by scheduling a fresh read.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class);
-        when(plugin.deleteTopology("foo")).thenReturn(CompletableFuture.completedFuture(null));
-        when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()))
-            .thenReturn(List.of(CompletableFuture.completedFuture(Map.of("foo", 4))));
-        when(runtime.scheduleWriteOperation(eq("clear-stored-topology-epoch"), eq(GROUP_TP), any()))
-            .thenReturn(CompletableFuture.completedFuture(null));
-
-        GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true);
-        service.runStreamsGroupTopologyCleanupCycle();
-        // Same runtime, second invocation: must schedule another read (flag released).
-        service.runStreamsGroupTopologyCleanupCycle();
-
-        verify(runtime, times(2)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any());
-    }
-
-    @Test
-    public void testShutdownCancelsScheduledCleanupTask() throws Exception {
-        // startup() with a plugin configured schedules the periodic cleanup tick; shutdown()
-        // must cancel that snapshot so the timer queue does not retain a self-rescheduling
-        // task referencing a torn-down runtime. MockTimer.size() filters cancelled entries,
-        // so observing 1 → 0 confirms the cancel() landed; advancing the clock past the
-        // interval afterwards must not fire the task — both the queue-skip on cancellation
-        // and the TimerTask body's isActive==false defensive return guard against it.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class);
-        MockTimer timer = new MockTimer();
-        GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true, timer);
-
-        assertEquals(1, timer.size());
-
-        service.shutdown();
-
-        assertEquals(0, timer.size());
-        // No cycle should fire even with the clock advanced well past the cleanup interval.
-        timer.advanceClock(Duration.ofHours(2).toMillis());
-        verify(runtime, never()).scheduleReadAllOperation(
-            eq("list-streams-groups-needing-topology-cleanup"), any());
-    }
-
-    @Test
-    public void testShutdownSafeWhenNoCleanupTaskScheduled() {
-        // Plugin absent => startup() short-circuits before scheduling a tick, so the
-        // cleanupTask field stays null. shutdown() must tolerate the null snapshot without
-        // throwing — broker close paths must not propagate.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        MockTimer timer = new MockTimer();
-        GroupCoordinatorService service = buildService(runtime, Optional.empty(), true, timer);
-
-        assertEquals(0, timer.size());
-        service.shutdown();
-        assertEquals(0, timer.size());
-    }
-
-    @Test
-    public void testCleanupCycleSingleFlightReleasesFlagOnEmptyPartitionList() {
-        // Pathological boundary: zero hosted partitions (e.g. broker just started, nothing
-        // loaded yet). partitionFutures is empty -> allOf(empty) -> immediate completion ->
-        // whenComplete still fires and releases the flag. A subsequent cycle must run.
-        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
-        StreamsGroupTopologyDescriptionPlugin plugin = mock(StreamsGroupTopologyDescriptionPlugin.class);
-        when(runtime.scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any()))
-            .thenReturn(List.of());
-
-        GroupCoordinatorService service = buildService(runtime, Optional.of(plugin), true);
-        service.runStreamsGroupTopologyCleanupCycle();
-        service.runStreamsGroupTopologyCleanupCycle();
-
-        verify(runtime, times(2)).scheduleReadAllOperation(eq("list-streams-groups-needing-topology-cleanup"), any());
-    }
-
    private static StreamsGroupTopologyDescriptionUpdateRequestData validUpdateRequest() {
        return new StreamsGroupTopologyDescriptionUpdateRequestData()
            .setGroupId("foo")
@@ -2899,169 +2899,6 @@ public class GroupCoordinatorShardTest {
        assertEquals(result, coordinator.fetchOffsets(request, Long.MAX_VALUE));
    }

-
-    @Test
-    public void testListStreamsGroupsNeedingTopologyCleanupOnlyReturnsEligible() {
-        GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
-        OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
-        GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
-            new LogContext(),
-            groupMetadataManager,
-            offsetMetadataManager,
-            new MockTime(),
-            new MockCoordinatorTimer<>(Time.SYSTEM),
-            mock(GroupCoordinatorConfig.class),
-            mock(CoordinatorMetrics.class),
-            mock(CoordinatorMetricsShard.class)
-        );
-
-        // Group 1: not empty — must be skipped.
-        StreamsGroup nonEmpty = mock(StreamsGroup.class);
-        when(nonEmpty.type()).thenReturn(Group.GroupType.STREAMS);
-        when(nonEmpty.isEmpty()).thenReturn(false);
-        when(nonEmpty.storedDescriptionTopologyEpoch(ArgumentMatchers.anyLong())).thenReturn(5);
-
-        // Group 2: empty but StoredDescriptionTopologyEpoch == -1 — must be skipped (already cleared).
-        StreamsGroup emptyNoStored = mock(StreamsGroup.class);
-        when(emptyNoStored.type()).thenReturn(Group.GroupType.STREAMS);
-        when(emptyNoStored.isEmpty()).thenReturn(true);
-        when(emptyNoStored.storedDescriptionTopologyEpoch(ArgumentMatchers.anyLong())).thenReturn(-1);
-
-        // Group 3: empty + stored, but offsets not yet expired — must be skipped.
-        StreamsGroup emptyStoredFreshOffsets = mock(StreamsGroup.class);
-        when(emptyStoredFreshOffsets.type()).thenReturn(Group.GroupType.STREAMS);
-        when(emptyStoredFreshOffsets.isEmpty()).thenReturn(true);
-        when(emptyStoredFreshOffsets.storedDescriptionTopologyEpoch(ArgumentMatchers.anyLong())).thenReturn(2);
-
-        // Group 4: empty + stored + all offsets expired — the only eligible group.
-        StreamsGroup emptyStoredExpired = mock(StreamsGroup.class);
-        when(emptyStoredExpired.type()).thenReturn(Group.GroupType.STREAMS);
-        when(emptyStoredExpired.isEmpty()).thenReturn(true);
-        when(emptyStoredExpired.storedDescriptionTopologyEpoch(ArgumentMatchers.anyLong())).thenReturn(7);
-
-        when(groupMetadataManager.groupIds()).thenReturn(Set.of(
-            "non-empty-streams", "empty-no-stored", "empty-stored-fresh-offsets", "empty-stored-expired"
-        ));
-        when(groupMetadataManager.group(eq("non-empty-streams"), ArgumentMatchers.anyLong())).thenReturn(nonEmpty);
-        when(groupMetadataManager.group(eq("empty-no-stored"), ArgumentMatchers.anyLong())).thenReturn(emptyNoStored);
-        when(groupMetadataManager.group(eq("empty-stored-fresh-offsets"), ArgumentMatchers.anyLong())).thenReturn(emptyStoredFreshOffsets);
-        when(groupMetadataManager.group(eq("empty-stored-expired"), ArgumentMatchers.anyLong())).thenReturn(emptyStoredExpired);
-        when(offsetMetadataManager.allOffsetsExpired(anyString(), ArgumentMatchers.anyLong())).thenReturn(true);
-        when(offsetMetadataManager.allOffsetsExpired(eq("empty-stored-fresh-offsets"), ArgumentMatchers.anyLong())).thenReturn(false);
-
-        Map<String, Integer> eligible = coordinator.listStreamsGroupsNeedingTopologyCleanup(0L);
-        assertEquals(Map.of("empty-stored-expired", 7), eligible);
-    }
-
-    @Test
-    public void testListStreamsGroupsNeedingTopologyCleanupContinuesOnUnexpectedError() {
-        GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
-        OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
-        GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
-            new LogContext(),
-            groupMetadataManager,
-            offsetMetadataManager,
-            new MockTime(),
-            new MockCoordinatorTimer<>(Time.SYSTEM),
-            mock(GroupCoordinatorConfig.class),
-            mock(CoordinatorMetrics.class),
-            mock(CoordinatorMetricsShard.class)
-        );
-
-        // Eligible group — should be returned despite the unrelated bad group throwing.
-        StreamsGroup good = mock(StreamsGroup.class);
-        when(good.type()).thenReturn(Group.GroupType.STREAMS);
-        when(good.isEmpty()).thenReturn(true);
-        when(good.storedDescriptionTopologyEpoch(ArgumentMatchers.anyLong())).thenReturn(4);
-
-        when(groupMetadataManager.groupIds()).thenReturn(Set.of("bad", "good"));
-        when(groupMetadataManager.group(eq("good"), ArgumentMatchers.anyLong())).thenReturn(good);
-        // "bad" surfaces a non-GroupIdNotFoundException. The broad catch must absorb it and the loop must continue.
-        when(groupMetadataManager.group(eq("bad"), ArgumentMatchers.anyLong()))
-            .thenThrow(new RuntimeException("unexpected!"));
-        when(offsetMetadataManager.allOffsetsExpired(eq("good"), ArgumentMatchers.anyLong())).thenReturn(true);
-
-        Map<String, Integer> eligible = coordinator.listStreamsGroupsNeedingTopologyCleanup(0L);
-        assertEquals(Map.of("good", 4), eligible);
-    }
-
-    @Test
-    public void testClearStoredDescriptionTopologyEpochIfMatchesMatched() {
-        GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
-        OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
-        GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
-            new LogContext(),
-            groupMetadataManager,
-            offsetMetadataManager,
-            Time.SYSTEM,
-            new MockCoordinatorTimer<>(Time.SYSTEM),
-            mock(GroupCoordinatorConfig.class),
-            mock(CoordinatorMetrics.class),
-            mock(CoordinatorMetricsShard.class)
-        );
-
-        StreamsGroup group = mock(StreamsGroup.class);
-        when(group.storedDescriptionTopologyEpoch()).thenReturn(7);
-        when(groupMetadataManager.group("g")).thenReturn(group);
-
-        CoordinatorRecord record = mock(CoordinatorRecord.class);
-        when(groupMetadataManager.updateStreamsGroupTopologyFields("g", -1, null))
-            .thenReturn(new CoordinatorResult<>(List.of(record)));
-
-        CoordinatorResult<Void, CoordinatorRecord> result = coordinator.clearStoredDescriptionTopologyEpochIfMatches("g", 7);
-        assertEquals(List.of(record), result.records());
-        verify(groupMetadataManager, times(1)).updateStreamsGroupTopologyFields("g", -1, null);
-    }
-
-    @Test
-    public void testClearStoredDescriptionTopologyEpochIfMatchesMismatchIsNoOp() {
-        GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
-        OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
-        GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
-            new LogContext(),
-            groupMetadataManager,
-            offsetMetadataManager,
-            Time.SYSTEM,
-            new MockCoordinatorTimer<>(Time.SYSTEM),
-            mock(GroupCoordinatorConfig.class),
-            mock(CoordinatorMetrics.class),
-            mock(CoordinatorMetricsShard.class)
-        );
-
-        StreamsGroup group = mock(StreamsGroup.class);
-        when(group.storedDescriptionTopologyEpoch()).thenReturn(8);
-        when(groupMetadataManager.group("g")).thenReturn(group);
-
-        CoordinatorResult<Void, CoordinatorRecord> result = coordinator.clearStoredDescriptionTopologyEpochIfMatches("g", 7);
-        assertTrue(result.records().isEmpty());
-        verify(groupMetadataManager, times(0)).updateStreamsGroupTopologyFields(
-            anyString(), ArgumentMatchers.any(), ArgumentMatchers.any());
-    }
-
-    @Test
-    public void testClearStoredDescriptionTopologyEpochIfMatchesUnknownGroupIsNoOp() {
-        GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
-        OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
-        GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
-            new LogContext(),
-            groupMetadataManager,
-            offsetMetadataManager,
-            Time.SYSTEM,
-            new MockCoordinatorTimer<>(Time.SYSTEM),
-            mock(GroupCoordinatorConfig.class),
-            mock(CoordinatorMetrics.class),
-            mock(CoordinatorMetricsShard.class)
-        );
-
-        when(groupMetadataManager.group("unknown"))
-            .thenThrow(new GroupIdNotFoundException("Group unknown not found."));
-
-        CoordinatorResult<Void, CoordinatorRecord> result = coordinator.clearStoredDescriptionTopologyEpochIfMatches("unknown", 7);
-        assertTrue(result.records().isEmpty());
-        verify(groupMetadataManager, times(0)).updateStreamsGroupTopologyFields(
-            anyString(), ArgumentMatchers.any(), ArgumentMatchers.any());
-    }
-
    public static class StubPlugin implements StreamsGroupTopologyDescriptionPlugin {
        @Override
        public void configure(Map<String, ?> configs) { }
@@ -920,12 +920,6 @@ public class StreamsConfig extends AbstractConfig {
            "Whether to use the configured <code>" + PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG + "</code> during global store/KTable processing. " +
                    "Disabled by default. This config will be removed in Kafka Streams 5.0, where global exception handling will be enabled by default";

-    /** {@code topology.description.push.enabled} */
-    public static final String TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG = "topology.description.push.enabled";
-    private static final String TOPOLOGY_DESCRIPTION_PUSH_ENABLED_DOC = "Controls whether the Kafka Streams client sends topology descriptions to the broker when requested. " +
-        "When set to false, the client will not prepare or push topology descriptions. " +
-        "Enabled by default.";
-
    static {
        CONFIG = new ConfigDef()

@@ -1378,12 +1372,7 @@ public class StreamsConfig extends AbstractConfig {
                    Type.LONG,
                    null,
                    Importance.LOW,
-                    WINDOW_SIZE_MS_DOC)
-            .define(TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG,
-                    Type.BOOLEAN,
-                    true,
-                    Importance.MEDIUM,
-                    TOPOLOGY_DESCRIPTION_PUSH_ENABLED_DOC);
+                    WINDOW_SIZE_MS_DOC);
    }

    // this is the list of configs for underlying clients