This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 9391bdd4f41 KAFKA-20625: Add
StreamsGroupTopologyDescriptionRequestManager and Streams client topology push
[1/N] (#22639)
9391bdd4f41 is described below
commit 9391bdd4f4179f06a77191d5ce05b42d4a43f466
Author: Lucy Liu <[email protected]>
AuthorDate: Tue Jun 23 16:46:54 2026 -0500
KAFKA-20625: Add StreamsGroupTopologyDescriptionRequestManager and Streams
client topology push [1/N] (#22639)
## Summary
Adds client-side logic to convert a Kafka Streams topology to wire
format and stash it on `StreamsRebalanceData` at `StreamThread` startup.
## File changes
- **`StreamsRebalanceData`**: three new fields
`wireTopologyDescription`, `memberId`, `topologyPushRequired` with
setters/getters and a unit test.
- **`StreamsConfig`**: new client config
`topology.description.push.enabled` (boolean, default `true`).
- **`TopologyDescriptionConverter`**: new internal helper that converts
`TopologyDescription` to
`StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription`,
with 4 unit tests covering multi-subtopology, branching, stateful
processor + global store, and dynamic sink topic.
- **`StreamThread.initStreamsRebalanceData`**: calls the converter at
startup and stashes the result on `StreamsRebalanceData` when the config
is enabled.
Reviewers: Lucas Brutschy <[email protected]>
---
.../consumer/internals/StreamsRebalanceData.java | 20 ++
.../internals/StreamsRebalanceDataTest.java | 27 ++
.../org/apache/kafka/streams/StreamsConfig.java | 13 +-
.../internals/InternalTopologyBuilder.java | 12 +
.../streams/processor/internals/StreamThread.java | 14 +-
.../internals/TopologyDescriptionConverter.java | 109 ++++++
.../processor/internals/StreamThreadTest.java | 4 +
.../TopologyDescriptionConverterTest.java | 366 +++++++++++++++++++++
8 files changed, 563 insertions(+), 2 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
index 286bcdb540c..e1b59a964b2 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
@@ -18,6 +18,7 @@ package org.apache.kafka.clients.consumer.internals;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
import java.util.ArrayList;
import java.util.Collection;
@@ -354,6 +355,10 @@ public class StreamsRebalanceData {
private final AtomicLong acceptableRecoveryLag = new AtomicLong(-1);
+ private final
AtomicReference<StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription>
wireTopologyDescription = new AtomicReference<>(null);
+
+ private final AtomicBoolean topologyPushRequired = new
AtomicBoolean(false);
+
public StreamsRebalanceData(final UUID processId,
final Optional<HostInfo> endpoint,
final Optional<String> rackId,
@@ -474,4 +479,19 @@ public class StreamsRebalanceData {
return acceptableRecoveryLag.get();
}
+ public void setWireTopologyDescription(final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wireDescription) {
+ wireTopologyDescription.set(wireDescription);
+ }
+
+ public
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
wireTopologyDescription() {
+ return wireTopologyDescription.get();
+ }
+
+ public void setTopologyPushRequired(final boolean topologyPushRequired) {
+ this.topologyPushRequired.set(topologyPushRequired);
+ }
+
+ public boolean topologyPushRequired() {
+ return topologyPushRequired.get();
+ }
}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java
index 92126a224d7..ea56a1fe6fa 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
@@ -32,6 +34,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -532,4 +536,27 @@ public class StreamsRebalanceDataTest {
assertEquals(1000, streamsRebalanceData.heartbeatIntervalMs());
}
+ @Test
+ public void streamsRebalanceDataShouldDefaultAndUpdateTopologyPushFields()
{
+ final StreamsRebalanceData streamsRebalanceData = new
StreamsRebalanceData(
+ UUID.randomUUID(),
+ Optional.of(new StreamsRebalanceData.HostInfo("localhost",
9090)),
+ Optional.empty(),
+ Map.of(),
+ Map.of("clientTag1", "clientTagValue1"),
+ Map::of
+ );
+
+ assertNull(streamsRebalanceData.wireTopologyDescription());
+ assertFalse(streamsRebalanceData.topologyPushRequired());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription();
+ streamsRebalanceData.setWireTopologyDescription(wire);
+ streamsRebalanceData.setTopologyPushRequired(true);
+
+ assertSame(wire, streamsRebalanceData.wireTopologyDescription());
+ assertTrue(streamsRebalanceData.topologyPushRequired());
+ }
+
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index d6099f4cec8..54dcb726c47 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -912,6 +912,12 @@ 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()
@@ -1359,7 +1365,12 @@ public class StreamsConfig extends AbstractConfig {
Type.LONG,
null,
Importance.LOW,
- WINDOW_SIZE_MS_DOC);
+ WINDOW_SIZE_MS_DOC)
+ .define(TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG,
+ Type.BOOLEAN,
+ true,
+ Importance.MEDIUM,
+ TOPOLOGY_DESCRIPTION_PUSH_ENABLED_DOC);
}
// this is the list of configs for underlying clients
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
index a19bbcc50eb..5b8e2642dcb 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
@@ -1457,6 +1457,18 @@ public class InternalTopologyBuilder {
return decorateTopic(topic);
}
+ /**
+ * If {@code topic} is an internally-managed topic in this topology (a
repartition topic,
+ * a changelog topic, etc.), return the decorated name with the
applicationId prefix.
+ * Otherwise return the topic name unchanged.
+ */
+ public String maybeDecorateInternalTopic(final String topic) {
+ if (topic != null &&
internalTopicNamesWithProperties.containsKey(topic)) {
+ return decorateTopic(topic);
+ }
+ return topic;
+ }
+
@SuppressWarnings("deprecation")
private String decorateTopic(final String topic) {
if (applicationId == null) {
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index 0433d7fb625..fb84041fe41 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -58,6 +58,7 @@ import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsConfig.InternalConfig;
import org.apache.kafka.streams.TaskMetadata;
import org.apache.kafka.streams.ThreadMetadata;
+import org.apache.kafka.streams.TopologyDescription;
import org.apache.kafka.streams.errors.MissingSourceTopicException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskCorruptedException;
@@ -706,7 +707,7 @@ public class StreamThread extends Thread implements
ProcessingThread {
final Map<String, StreamsRebalanceData.Subtopology> subtopologies =
initBrokerTopology(config, internalTopologyBuilder);
- return new StreamsRebalanceData(
+ final StreamsRebalanceData streamsRebalanceData = new
StreamsRebalanceData(
processId,
endpoint,
rackId,
@@ -714,6 +715,17 @@ public class StreamThread extends Thread implements
ProcessingThread {
config.getClientTags(),
taskOffsetSum
);
+
+ if
(config.getBoolean(StreamsConfig.TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG)) {
+ final TopologyDescription description =
internalTopologyBuilder.describe();
+ streamsRebalanceData.setWireTopologyDescription(
+ TopologyDescriptionConverter.toWire(
+ description,
+ internalTopologyBuilder::maybeDecorateInternalTopic)
+ );
+ }
+
+ return streamsRebalanceData;
}
private static Map<String, StreamsRebalanceData.Subtopology>
initBrokerTopology(final StreamsConfig config,
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyDescriptionConverter.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyDescriptionConverter.java
new file mode 100644
index 00000000000..5c71ad725df
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyDescriptionConverter.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import org.apache.kafka.streams.TopologyDescription;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public final class TopologyDescriptionConverter {
+
+ private static final byte NODE_TYPE_SOURCE = 1;
+ private static final byte NODE_TYPE_PROCESSOR = 2;
+ private static final byte NODE_TYPE_SINK = 3;
+
+ private TopologyDescriptionConverter() {}
+
+ public static
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription
toWire(final TopologyDescription description,
+
final Function<String, String> topicNameDecorator) {
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription();
+
+ wire.setSubtopologies(description.subtopologies().stream()
+
.sorted(Comparator.comparingInt(TopologyDescription.Subtopology::id))
+ .map(s -> toWireSubtopology(s, topicNameDecorator))
+ .collect(Collectors.toList()));
+
+ wire.setGlobalStores(description.globalStores().stream()
+
.sorted(Comparator.comparingInt(TopologyDescription.GlobalStore::id))
+ .map(g -> toWireGlobalStore(g, topicNameDecorator))
+ .collect(Collectors.toList()));
+
+ return wire;
+ }
+
+ private static
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology
toWireSubtopology(
+ final TopologyDescription.Subtopology subtopology,
+ final Function<String, String> topicNameDecorator) {
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology
wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology();
+ wire.setSubtopologyId(String.valueOf(subtopology.id()));
+ wire.setNodes(subtopology.nodes().stream()
+ .sorted(Comparator.comparing(TopologyDescription.Node::name))
+ .map(n -> toWireNode(n, topicNameDecorator))
+ .collect(Collectors.toList()));
+ return wire;
+ }
+
+ private static
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
toWireNode(
+ final TopologyDescription.Node node,
+ final Function<String, String> topicNameDecorator) {
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode();
+ wire.setName(node.name());
+ wire.setSuccessors(node.successors().stream()
+ .map(TopologyDescription.Node::name)
+ .sorted()
+ .collect(Collectors.toList()));
+
+ if (node instanceof TopologyDescription.Source) {
+ final TopologyDescription.Source source =
(TopologyDescription.Source) node;
+ wire.setNodeType(NODE_TYPE_SOURCE);
+ wire.setSourceTopics(source.topicSet() == null
+ ? new ArrayList<>()
+ : source.topicSet().stream()
+ .map(topicNameDecorator)
+ .sorted()
+ .collect(Collectors.toList()));
+ } else if (node instanceof TopologyDescription.Sink) {
+ final TopologyDescription.Sink sink = (TopologyDescription.Sink)
node;
+ wire.setNodeType(NODE_TYPE_SINK);
+ wire.setSinkTopic(topicNameDecorator.apply(sink.topic()));
+ } else if (node instanceof TopologyDescription.Processor) {
+ final TopologyDescription.Processor processor =
(TopologyDescription.Processor) node;
+ wire.setNodeType(NODE_TYPE_PROCESSOR);
+
wire.setStores(processor.stores().stream().sorted().collect(Collectors.toList()));
+ } else {
+ throw new IllegalStateException("Unknown node type: " +
node.getClass().getName());
+ }
+ return wire;
+ }
+
+ private static
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore
toWireGlobalStore(
+ final TopologyDescription.GlobalStore globalStore,
+ final Function<String, String> topicNameDecorator) {
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore
wire =
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore();
+ wire.setSource(toWireNode(globalStore.source(), topicNameDecorator));
+ wire.setProcessor(toWireNode(globalStore.processor(),
topicNameDecorator));
+ return wire;
+ }
+}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index dbff3d6933f..22415ea0225 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -3854,6 +3854,10 @@ public class StreamThreadTest {
)
));
when(topologyBuilder.copartitionGroups()).thenReturn(Set.of(Set.of("source1")));
+ final InternalTopologyBuilder.TopologyDescription mockDescription =
mock(InternalTopologyBuilder.TopologyDescription.class);
+
when(mockDescription.subtopologies()).thenReturn(Collections.emptySet());
+
when(mockDescription.globalStores()).thenReturn(Collections.emptySet());
+ when(topologyBuilder.describe()).thenReturn(mockDescription);
final StreamsMetricsImpl streamsMetrics = new StreamsMetricsImpl(
metrics,
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TopologyDescriptionConverterTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TopologyDescriptionConverterTest.java
new file mode 100644
index 00000000000..c65b7481cb1
--- /dev/null
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TopologyDescriptionConverterTest.java
@@ -0,0 +1,366 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import org.apache.kafka.streams.TopologyDescription;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TopologyDescriptionConverterTest {
+
+ /**
+ * Test the situation when the topology has multiple subtopologies,
including one with two
+ * source nodes feeding the same processor: subtopologies should be sorted
by id and each node
+ * should be emitted with its correct type, topics, and successors.
+ */
+ @Test
+ public void shouldConvertMultipleSubtopologiesWithMultiSourceProcessor() {
+ final TopologyDescription.Source source0 =
mock(TopologyDescription.Source.class);
+ final TopologyDescription.Sink sink0 =
mock(TopologyDescription.Sink.class);
+ when(source0.name()).thenReturn("source-0");
+ when(source0.topicSet()).thenReturn(Set.of("topic-0-in"));
+ when(source0.successors()).thenReturn(Set.of(sink0));
+ when(sink0.name()).thenReturn("sink-0");
+ when(sink0.topic()).thenReturn("topic-0-out");
+ when(sink0.successors()).thenReturn(Set.of());
+
+ final TopologyDescription.Subtopology subtopology0 =
mock(TopologyDescription.Subtopology.class);
+ when(subtopology0.id()).thenReturn(0);
+ when(subtopology0.nodes()).thenReturn(Set.of(source0, sink0));
+
+ final TopologyDescription.Source source1a =
mock(TopologyDescription.Source.class);
+ final TopologyDescription.Source source1b =
mock(TopologyDescription.Source.class);
+ final TopologyDescription.Processor processor1 =
mock(TopologyDescription.Processor.class);
+ final TopologyDescription.Sink sink1 =
mock(TopologyDescription.Sink.class);
+
+ when(source1a.name()).thenReturn("source-1-a");
+ when(source1a.topicSet()).thenReturn(Set.of("topic-1-z", "topic-1-a"));
+ when(source1a.successors()).thenReturn(Set.of(processor1));
+
+ when(source1b.name()).thenReturn("source-1-b");
+ when(source1b.topicSet()).thenReturn(Set.of("topic-1-m"));
+ when(source1b.successors()).thenReturn(Set.of(processor1));
+
+ when(processor1.name()).thenReturn("processor-1");
+ when(processor1.stores()).thenReturn(Set.of());
+ when(processor1.successors()).thenReturn(Set.of(sink1));
+
+ when(sink1.name()).thenReturn("sink-1");
+ when(sink1.topic()).thenReturn("topic-1-out");
+ when(sink1.successors()).thenReturn(Set.of());
+
+ final TopologyDescription.Subtopology subtopology1 =
mock(TopologyDescription.Subtopology.class);
+ when(subtopology1.id()).thenReturn(1);
+ when(subtopology1.nodes()).thenReturn(Set.of(source1a, source1b,
processor1, sink1));
+
+ final TopologyDescription description =
mock(TopologyDescription.class);
+ when(description.subtopologies()).thenReturn(Set.of(subtopology1,
subtopology0));
+ when(description.globalStores()).thenReturn(Set.of());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription wire =
+ TopologyDescriptionConverter.toWire(description,
Function.identity());
+
+ assertTrue(wire.globalStores().isEmpty());
+ assertEquals(2, wire.subtopologies().size());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology
wireSub0 = wire.subtopologies().get(0);
+ assertEquals("0", wireSub0.subtopologyId());
+ assertEquals(2, wireSub0.nodes().size());
+ assertEquals("sink-0", wireSub0.nodes().get(0).name());
+ assertEquals("source-0", wireSub0.nodes().get(1).name());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology
wireSub1 = wire.subtopologies().get(1);
+ assertEquals("1", wireSub1.subtopologyId());
+ assertEquals(4, wireSub1.nodes().size());
+
+ final
List<StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode>
nodes = wireSub1.nodes();
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireProcessor1 = nodes.get(0);
+ assertEquals("processor-1", wireProcessor1.name());
+ assertEquals((byte) 2, wireProcessor1.nodeType());
+ assertEquals(List.of(), wireProcessor1.stores());
+ assertEquals(List.of("sink-1"), wireProcessor1.successors());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSink1 = nodes.get(1);
+ assertEquals("sink-1", wireSink1.name());
+ assertEquals((byte) 3, wireSink1.nodeType());
+ assertEquals("topic-1-out", wireSink1.sinkTopic());
+ assertEquals(List.of(), wireSink1.successors());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSource1a = nodes.get(2);
+ assertEquals("source-1-a", wireSource1a.name());
+ assertEquals((byte) 1, wireSource1a.nodeType());
+ assertEquals(List.of("topic-1-a", "topic-1-z"),
wireSource1a.sourceTopics());
+ assertEquals(List.of("processor-1"), wireSource1a.successors());
+ assertNull(wireSource1a.sinkTopic());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSource1b = nodes.get(3);
+ assertEquals("source-1-b", wireSource1b.name());
+ assertEquals((byte) 1, wireSource1b.nodeType());
+ assertEquals(List.of("topic-1-m"), wireSource1b.sourceTopics());
+ assertEquals(List.of("processor-1"), wireSource1b.successors());
+ }
+
+ /**
+ * Test the situation when a single source node branches into multiple
processors: the source's
+ * successors list should contain all branch processors sorted
alphabetically.
+ */
+ @Test
+ public void shouldConvertBranchTopology() {
+ final TopologyDescription.Source source =
mock(TopologyDescription.Source.class);
+ final TopologyDescription.Processor processor1 =
mock(TopologyDescription.Processor.class);
+ final TopologyDescription.Processor processor2 =
mock(TopologyDescription.Processor.class);
+ final TopologyDescription.Sink sink =
mock(TopologyDescription.Sink.class);
+
+ when(source.name()).thenReturn("source");
+ when(source.topicSet()).thenReturn(Set.of("input-topic"));
+ when(source.successors()).thenReturn(Set.of(processor1, processor2));
+
+ when(processor1.name()).thenReturn("processor-1");
+ when(processor1.stores()).thenReturn(Set.of());
+ when(processor1.successors()).thenReturn(Set.of(sink));
+
+ when(processor2.name()).thenReturn("processor-2");
+ when(processor2.stores()).thenReturn(Set.of());
+ when(processor2.successors()).thenReturn(Set.of(sink));
+
+ when(sink.name()).thenReturn("sink");
+ when(sink.topic()).thenReturn("output-topic");
+ when(sink.successors()).thenReturn(Set.of());
+
+ final TopologyDescription.Subtopology subtopology =
mock(TopologyDescription.Subtopology.class);
+ when(subtopology.id()).thenReturn(0);
+ when(subtopology.nodes()).thenReturn(Set.of(source, processor1,
processor2, sink));
+
+ final TopologyDescription description =
mock(TopologyDescription.class);
+ when(description.subtopologies()).thenReturn(Set.of(subtopology));
+ when(description.globalStores()).thenReturn(Set.of());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription wire =
+ TopologyDescriptionConverter.toWire(description,
Function.identity());
+
+ assertTrue(wire.globalStores().isEmpty());
+ assertEquals(1, wire.subtopologies().size());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology
wireSub = wire.subtopologies().get(0);
+ assertEquals("0", wireSub.subtopologyId());
+ assertEquals(4, wireSub.nodes().size());
+
+ final
List<StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode>
nodes = wireSub.nodes();
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireProcessor1 = nodes.get(0);
+ assertEquals("processor-1", wireProcessor1.name());
+ assertEquals((byte) 2, wireProcessor1.nodeType());
+ assertEquals(List.of(), wireProcessor1.stores());
+ assertEquals(List.of("sink"), wireProcessor1.successors());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireProcessor2 = nodes.get(1);
+ assertEquals("processor-2", wireProcessor2.name());
+ assertEquals((byte) 2, wireProcessor2.nodeType());
+ assertEquals(List.of(), wireProcessor2.stores());
+ assertEquals(List.of("sink"), wireProcessor2.successors());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSink = nodes.get(2);
+ assertEquals("sink", wireSink.name());
+ assertEquals((byte) 3, wireSink.nodeType());
+ assertEquals("output-topic", wireSink.sinkTopic());
+ assertEquals(List.of(), wireSink.successors());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSource = nodes.get(3);
+ assertEquals("source", wireSource.name());
+ assertEquals((byte) 1, wireSource.nodeType());
+ assertEquals(List.of("input-topic"), wireSource.sourceTopics());
+ assertEquals(List.of("processor-1", "processor-2"),
wireSource.successors());
+ assertNull(wireSource.sinkTopic());
+ }
+
+ /**
+ * Test the situation when the topology contains a stateful processor and
a global store: state
+ * store names should appear on the processor, and the global store should
be emitted under
+ * globalStores rather than subtopologies.
+ */
+ @Test
+ public void shouldConvertTopologyWithStatefulProcessorAndGlobalStore() {
+ final TopologyDescription.Source source =
mock(TopologyDescription.Source.class);
+ final TopologyDescription.Processor processor =
mock(TopologyDescription.Processor.class);
+ final TopologyDescription.Sink sink =
mock(TopologyDescription.Sink.class);
+
+ when(source.name()).thenReturn("source");
+ when(source.topicSet()).thenReturn(Set.of("input-topic"));
+ when(source.successors()).thenReturn(Set.of(processor));
+
+ when(processor.name()).thenReturn("processor");
+ when(processor.stores()).thenReturn(Set.of("store-1", "store-2"));
+ when(processor.successors()).thenReturn(Set.of(sink));
+
+ when(sink.name()).thenReturn("sink");
+ when(sink.topic()).thenReturn("output-topic");
+ when(sink.successors()).thenReturn(Set.of());
+
+ final TopologyDescription.Source globalSource =
mock(TopologyDescription.Source.class);
+ when(globalSource.name()).thenReturn("global-source");
+ when(globalSource.topicSet()).thenReturn(Set.of("global-topic"));
+ when(globalSource.successors()).thenReturn(Set.of());
+
+ final TopologyDescription.Processor globalProcessor =
mock(TopologyDescription.Processor.class);
+ when(globalProcessor.name()).thenReturn("global-processor");
+ when(globalProcessor.stores()).thenReturn(Set.of("global-store"));
+ when(globalProcessor.successors()).thenReturn(Set.of());
+
+ final TopologyDescription.GlobalStore globalStore =
mock(TopologyDescription.GlobalStore.class);
+ when(globalStore.id()).thenReturn(0);
+ when(globalStore.source()).thenReturn(globalSource);
+ when(globalStore.processor()).thenReturn(globalProcessor);
+
+ final TopologyDescription.Subtopology subtopology =
mock(TopologyDescription.Subtopology.class);
+ when(subtopology.id()).thenReturn(0);
+ when(subtopology.nodes()).thenReturn(Set.of(source, processor, sink));
+
+ final TopologyDescription description =
mock(TopologyDescription.class);
+ when(description.subtopologies()).thenReturn(Set.of(subtopology));
+ when(description.globalStores()).thenReturn(Set.of(globalStore));
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription wire =
+ TopologyDescriptionConverter.toWire(description,
Function.identity());
+
+ assertEquals(1, wire.subtopologies().size());
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionSubtopology
wireSub = wire.subtopologies().get(0);
+ assertEquals("0", wireSub.subtopologyId());
+ assertEquals(3, wireSub.nodes().size());
+
+ final
List<StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode>
nodes = wireSub.nodes();
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireProcessor = nodes.get(0);
+ assertEquals("processor", wireProcessor.name());
+ assertEquals((byte) 2, wireProcessor.nodeType());
+ assertEquals(List.of("store-1", "store-2"), wireProcessor.stores());
+ assertEquals(List.of("sink"), wireProcessor.successors());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSink = nodes.get(1);
+ assertEquals("sink", wireSink.name());
+ assertEquals((byte) 3, wireSink.nodeType());
+ assertEquals("output-topic", wireSink.sinkTopic());
+ assertEquals(List.of(), wireSink.successors());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSource = nodes.get(2);
+ assertEquals("source", wireSource.name());
+ assertEquals((byte) 1, wireSource.nodeType());
+ assertEquals(List.of("input-topic"), wireSource.sourceTopics());
+ assertEquals(List.of("processor"), wireSource.successors());
+ assertNull(wireSource.sinkTopic());
+
+ assertEquals(1, wire.globalStores().size());
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionGlobalStore
wireGlobal = wire.globalStores().get(0);
+ assertEquals("global-source", wireGlobal.source().name());
+ assertEquals((byte) 1, wireGlobal.source().nodeType());
+ assertEquals(List.of("global-topic"),
wireGlobal.source().sourceTopics());
+ assertEquals(List.of(), wireGlobal.source().successors());
+ assertEquals("global-processor", wireGlobal.processor().name());
+ assertEquals((byte) 2, wireGlobal.processor().nodeType());
+ assertEquals(List.of("global-store"), wireGlobal.processor().stores());
+ assertEquals(List.of(), wireGlobal.processor().successors());
+ }
+
+ /**
+ * Test the situation when a sink uses a TopicNameExtractor (dynamic
topic): the wire
+ * SinkTopic field should be null since the schema has no field for the
extractor itself.
+ */
+ @Test
+ public void shouldConvertSinkWithDynamicTopic() {
+ final TopologyDescription.Source source =
mock(TopologyDescription.Source.class);
+ final TopologyDescription.Sink sink =
mock(TopologyDescription.Sink.class);
+
+ when(source.name()).thenReturn("source");
+ when(source.topicSet()).thenReturn(Set.of("input-topic"));
+ when(source.successors()).thenReturn(Set.of(sink));
+
+ when(sink.name()).thenReturn("sink");
+ when(sink.topic()).thenReturn(null);
+ when(sink.successors()).thenReturn(Set.of());
+
+ final TopologyDescription.Subtopology subtopology =
mock(TopologyDescription.Subtopology.class);
+ when(subtopology.id()).thenReturn(0);
+ when(subtopology.nodes()).thenReturn(Set.of(source, sink));
+
+ final TopologyDescription description =
mock(TopologyDescription.class);
+ when(description.subtopologies()).thenReturn(Set.of(subtopology));
+ when(description.globalStores()).thenReturn(Set.of());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription wire =
+ TopologyDescriptionConverter.toWire(description,
Function.identity());
+
+ final
List<StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode>
nodes =
+ wire.subtopologies().get(0).nodes();
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSink = nodes.get(0);
+ assertEquals("sink", wireSink.name());
+ assertEquals((byte) 3, wireSink.nodeType());
+ assertNull(wireSink.sinkTopic());
+ }
+
+ /**
+ * Test the situation when a topic-name decorator is supplied: internal
topic names should be
+ * decorated with the applicationId prefix on the wire output, while
external (user-provided)
+ * topic names should pass through unchanged.
+ */
+ @Test
+ public void shouldApplyTopicNameDecorator() {
+ final TopologyDescription.Source source =
mock(TopologyDescription.Source.class);
+ final TopologyDescription.Sink sink =
mock(TopologyDescription.Sink.class);
+ when(source.name()).thenReturn("source");
+ when(source.topicSet()).thenReturn(Set.of("external-input-topic",
"repartition-topic"));
+ when(source.successors()).thenReturn(Set.of(sink));
+ when(sink.name()).thenReturn("sink");
+ when(sink.topic()).thenReturn("changelog-topic");
+ when(sink.successors()).thenReturn(Set.of());
+
+ final TopologyDescription.Subtopology subtopology =
mock(TopologyDescription.Subtopology.class);
+ when(subtopology.id()).thenReturn(0);
+ when(subtopology.nodes()).thenReturn(Set.of(source, sink));
+
+ final TopologyDescription description =
mock(TopologyDescription.class);
+ when(description.subtopologies()).thenReturn(Set.of(subtopology));
+ when(description.globalStores()).thenReturn(Set.of());
+
+ final Set<String> internalTopics = Set.of("repartition-topic",
"changelog-topic");
+ final Function<String, String> topicNameDecorator = name ->
+ internalTopics.contains(name) ? "my-app-" + name : name;
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription wire =
+ TopologyDescriptionConverter.toWire(description,
topicNameDecorator);
+
+ final
List<StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode>
nodes =
+ wire.subtopologies().get(0).nodes();
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSink = nodes.get(0);
+ assertEquals("my-app-changelog-topic", wireSink.sinkTopic());
+
+ final
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode
wireSource = nodes.get(1);
+ assertEquals(List.of("external-input-topic",
"my-app-repartition-topic"), wireSource.sourceTopics());
+ }
+}
