You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ab...@apache.org on 2022/10/07 04:33:28 UTC
[kafka] branch trunk updated: MINOR: Include all hosts in metadata for topology (#12594)
This is an automated email from the ASF dual-hosted git repository.
ableegoldman 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 cbdcd20ac18 MINOR: Include all hosts in metadata for topology (#12594)
cbdcd20ac18 is described below
commit cbdcd20ac183c8935161179e2d8cc53eabcb027c
Author: Walker Carlson <18...@users.noreply.github.com>
AuthorDate: Thu Oct 6 23:33:00 2022 -0500
MINOR: Include all hosts in metadata for topology (#12594)
When building streams metadata we want to build even if the host is empty as it is a common way to find the other host addresses
Reviewers: John Roesler <vv...@apache.org>, Anna Sophie Blee-Goldman <ab...@apache.org>
---
.../processor/internals/StreamsMetadataState.java | 28 ++++++++++------------
.../integration/NamedTopologyIntegrationTest.java | 12 ++++++++--
2 files changed, 22 insertions(+), 18 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java
index 6850715b0aa..61add951b22 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java
@@ -389,22 +389,18 @@ public class StreamsMetadataState {
standbyStoresOnHost.addAll(getStoresOnHost(storeToSourceTopics, standbyPartitionsOnHost));
}
- if (!(activeStoresOnHost.isEmpty() && activePartitionsOnHost.isEmpty() && standbyStoresOnHost.isEmpty() && standbyPartitionsOnHost.isEmpty())) {
- final StreamsMetadata metadata = new StreamsMetadataImpl(
- hostInfo,
- activeStoresOnHost,
- activePartitionsOnHost,
- standbyStoresOnHost,
- standbyPartitionsOnHost,
- topologyName
- );
-
- rebuiltMetadata.add(metadata);
- if (hostInfo.equals(thisHost)) {
- localMetadata.set(metadata);
- }
- } else {
- log.debug("Host {} has no tasks for topology {} at the moment, this metadata will not be built", hostInfo, topologyName);
+ final StreamsMetadata metadata = new StreamsMetadataImpl(
+ hostInfo,
+ activeStoresOnHost,
+ activePartitionsOnHost,
+ standbyStoresOnHost,
+ standbyPartitionsOnHost,
+ topologyName
+ );
+
+ rebuiltMetadata.add(metadata);
+ if (hostInfo.equals(thisHost)) {
+ localMetadata.set(metadata);
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java
index 8facc8279d2..5cc693265c6 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java
@@ -418,10 +418,14 @@ public class NamedTopologyIntegrationTest {
streams.streamsMetadataForStore(topology2Store, TOPOLOGY_2),
streams2.streamsMetadataForStore(topology2Store, TOPOLOGY_2));
+ assertThat(streams.allStreamsClientsMetadataForTopology(TOPOLOGY_1).size(), equalTo(2));
+ assertThat(streams2.allStreamsClientsMetadataForTopology(TOPOLOGY_1).size(), equalTo(2));
verifyMetadataForTopology(
TOPOLOGY_1,
streams.allStreamsClientsMetadataForTopology(TOPOLOGY_1),
streams2.allStreamsClientsMetadataForTopology(TOPOLOGY_1));
+ assertThat(streams.allStreamsClientsMetadataForTopology(TOPOLOGY_2).size(), equalTo(2));
+ assertThat(streams2.allStreamsClientsMetadataForTopology(TOPOLOGY_2).size(), equalTo(2));
verifyMetadataForTopology(
TOPOLOGY_2,
streams.allStreamsClientsMetadataForTopology(TOPOLOGY_2),
@@ -924,8 +928,12 @@ public class NamedTopologyIntegrationTest {
.collect(Collectors.toList())),
is(true));
- // then verify that only this topology's one store appears
- assertThat(metadata.stateStoreNames(), equalTo(singleton("store-" + topologyName)));
+ // then verify that only this topology's one store appears if the host has partitions assigned
+ if (!metadata.topicPartitions().isEmpty()) {
+ assertThat(metadata.stateStoreNames(), equalTo(singleton("store-" + topologyName)));
+ } else {
+ assertThat(metadata.stateStoreNames().isEmpty(), is(true));
+ }
// finally make sure the standby fields are empty since they are not enabled for this test
assertThat(metadata.standbyTopicPartitions().isEmpty(), is(true));