MINOR: Consider repartition source topic for changelog topics (#19373)

With the Streams rebalance protocol, when the partition count 
for changelog topics are computed the topic manager only
considers external source topics and throws if there are no
source topics. However, subtopologies are allowed to only
have repartition topics as source topics.

This commit also considers repartition source topics when
computing the partition count for changelog topics.

Reviewers: Lucas Brutschy <[email protected]>

Author: cadonna

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ChangelogTopics.java

@@ -31,6 +31,7 @@
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * This class is responsible for setting up the changelog topics for a topology. For a given topology, which does not have the number
@@ -75,7 +76,10 @@ public Map<String, Integer> setup() {
             final Set<String> sourceTopics = new HashSet<>(subtopology.sourceTopics());
 
             final OptionalInt maxNumPartitions =
-                subtopology.sourceTopics().stream().mapToInt(this::getPartitionCountOrFail).max();
+                Stream.concat(
+                    subtopology.sourceTopics().stream(),
+                    subtopology.repartitionSourceTopics().stream().map(TopicInfo::name)
+                ).mapToInt(this::getPartitionCountOrFail).max();
 
             if (maxNumPartitions.isEmpty()) {
                 throw new StreamsInvalidTopologyException("No source topics found for subtopology " + subtopology.subtopologyId());

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ChangelogTopicsTest.java

@@ -28,6 +28,7 @@
 import java.util.Map;
 import java.util.OptionalInt;
 
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -49,6 +50,18 @@ public class ChangelogTopicsTest {
         .setRepartitionSinkTopics(List.of(SINK_TOPIC_NAME))
         .setRepartitionSourceTopics(List.of(REPARTITION_TOPIC_INFO))
         .setStateChangelogTopics(List.of());
+    private static final Subtopology SUBTOPOLOGY_NO_REPARTITION_SOURCE = new Subtopology()
+        .setSubtopologyId("SUBTOPOLOGY_NO_SOURCE")
+        .setSourceTopics(List.of(SOURCE_TOPIC_NAME))
+        .setRepartitionSinkTopics(List.of(SINK_TOPIC_NAME))
+        .setRepartitionSourceTopics(List.of())
+        .setStateChangelogTopics(List.of());
+    private static final Subtopology SUBTOPOLOGY_NO_SOURCE_NO_REPARTITION_SOURCE = new Subtopology()
+        .setSubtopologyId("SUBTOPOLOGY_NO_SOURCE")
+        .setSourceTopics(List.of())
+        .setRepartitionSinkTopics(List.of(SINK_TOPIC_NAME))
+        .setRepartitionSourceTopics(List.of())
+        .setStateChangelogTopics(List.of());
     private static final Subtopology SUBTOPOLOGY_STATELESS = new Subtopology()
         .setSubtopologyId("SUBTOPOLOGY_STATELESS")
         .setSourceTopics(List.of(SOURCE_TOPIC_NAME))
@@ -85,8 +98,8 @@ private static OptionalInt topicPartitionProvider(String s) {
     }
 
     @Test
-    public void shouldFailIfNoSourceTopics() {
-        final List<Subtopology> subtopologies = List.of(SUBTOPOLOGY_NO_SOURCE);
+    public void shouldFailIfNoSourceTopicsAndNoRepartitionSourceTopics() {
+        final List<Subtopology> subtopologies = List.of(SUBTOPOLOGY_NO_SOURCE_NO_REPARTITION_SOURCE);
 
         final ChangelogTopics changelogTopics =
             new ChangelogTopics(LOG_CONTEXT, subtopologies, ChangelogTopicsTest::topicPartitionProvider);
@@ -95,6 +108,24 @@ public void shouldFailIfNoSourceTopics() {
         assertTrue(e.getMessage().contains("No source topics found for subtopology"));
     }
 
+    @Test
+    public void shouldNotFailIfOnlySourceTopicsEmpty() {
+        final List<Subtopology> subtopologies = List.of(SUBTOPOLOGY_NO_SOURCE);
+
+        final ChangelogTopics changelogTopics =
+            new ChangelogTopics(LOG_CONTEXT, subtopologies, ChangelogTopicsTest::topicPartitionProvider);
+        assertDoesNotThrow(changelogTopics::setup);
+    }
+
+    @Test
+    public void shouldNotFailIfOnlyRepartitionSourceTopicsEmpty() {
+        final List<Subtopology> subtopologies = List.of(SUBTOPOLOGY_NO_REPARTITION_SOURCE);
+
+        final ChangelogTopics changelogTopics =
+            new ChangelogTopics(LOG_CONTEXT, subtopologies, ChangelogTopicsTest::topicPartitionProvider);
+        assertDoesNotThrow(changelogTopics::setup);
+    }
+
     @Test
     public void shouldNotContainChangelogsForStatelessTasks() {
         final List<Subtopology> subtopologies = List.of(SUBTOPOLOGY_STATELESS);