This is an automated email from the ASF dual-hosted git repository.
GGraziadei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new b9f9b182a Deliver upstream feedback to remote tasks and stop spurious
"No remote destination available" warnings (#9175)
b9f9b182a is described below
commit b9f9b182a92c6fb333160fa088533e489f8b3645
Author: ErnestasVaiciukeviciusAdform <[email protected]>
AuthorDate: Fri Oct 23 20:59:02 2026 +0300
Deliver upstream feedback to remote tasks and stop spurious "No remote
destination available" warnings (#9175)
* Deliver upstream feedback to remote tasks and stop spurious "No remote
destination available" warnings
When topology.upstream.feedback.enable is true, feedback tuples are routed
directly to upstream (source) component tasks, but a worker's routable
remote
destinations (cachedTaskToNodePort) and connections are built only from
outboundTasks, which contained just the downstream-target components of
local
tasks. Feedback recipients were therefore missing, so TransferDrainer logged
"No remote destination available for task" and silently dropped the tuple;
cross-worker feedback was never delivered. Whether the warning appeared
depended
on task placement, which was misleading.
Include the non-system source (upstream) components of local tasks in
outboundTasks when feedback is enabled, mirroring
Executor.computeUpstreamTaskIds
at worker granularity, so feedback recipients are routable and connected.
Co-authored-by: Copilot <[email protected]>
* Remove unnecessary test, cleanup comments
* Upstream feedback setup: use topologyConf and exclude system components
* Enhance WorkerOutboundTasksTest: add tests for feedback routing and acker
task behavior
* Refactor WorkerState test utilities: enhance getWorkerState method and
add feedback configuration tests
* Apply suggestions from compiler warnings
* Move testTopologyConfigEnabledFeedback/testTopologyConfigDisabledFeedback
to WorkerOutboundTasksTest
* Improve testTopologyConfigDisabledFeedback testcase
* Refactor NoopSpout and NoopBolt classes: remove unnecessary interface
implementations and simplify method signatures
* Enhance documentation for upstream feedback: clarify connection overhead
and security implications in Config.java and Performance.md
* WorkerOutboundTasksTest: made test naming consistent
* New test cases for sink-only worker, with/without ackers
* Extracted common logic of upstream components for feedback from
WorkerState and Executor into StormCommon.upstreamFeedbackSourceComponents
---------
Co-authored-by: Copilot <[email protected]>
---
docs/Performance.md | 13 ++
storm-client/src/jvm/org/apache/storm/Config.java | 6 +
.../jvm/org/apache/storm/daemon/StormCommon.java | 19 ++
.../apache/storm/daemon/worker/WorkerState.java | 26 ++-
.../jvm/org/apache/storm/executor/Executor.java | 7 +-
.../org/apache/storm/daemon/StormCommonTest.java | 66 ++++++
.../daemon/worker/TestUtilsForWorkerState.java | 31 +--
.../daemon/worker/WorkerOutboundTasksTest.java | 254 +++++++++++++++++++++
8 files changed, 400 insertions(+), 22 deletions(-)
diff --git a/docs/Performance.md b/docs/Performance.md
index 5774a8876..393ae0fca 100644
--- a/docs/Performance.md
+++ b/docs/Performance.md
@@ -187,6 +187,19 @@ Impact of this on performance is dependent on the topology
and its deployment fo
Consequently it is useful to assess the impact of setting
`topology.disable.loadaware.messaging` to `true` or `false` for your
specific case.
+### Jitter-aware messaging
+
+`org.apache.storm.grouping.JitterAwareStreamGrouping` is an opt-in grouping
that favors downstream
+tasks reporting lower jitter, falling back to load-aware shuffle when jitter
feedback cannot
+distinguish the candidates. It requires both `topology.stats.ewma.enable=true`
and
+`topology.upstream.feedback.enable=true` (both default to `false`). Enabling
feedback alone
+does not select this grouping.
+
+Enabling upstream feedback increases inter-worker connection fan-in: each
worker hosting a user
+bolt establishes an outbound Netty client connection to every remote worker
hosting any task of
+that bolt's immediate upstream components (spouts or bolts), regardless of
stream grouping.
+Connections are shared across tasks targeting the same worker, and existing
outbound connections
+are reused. Same-worker feedback does not require a network connection.
## 7. Sampling Rate
Sampling rate is used to control how often certain metrics are computed on the
Spout and Bolt executors. This is configured using `topology.stats.sample.rate`
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index 557872754..de051f2f7 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -615,6 +615,12 @@ public class Config extends HashMap<String, Object> {
* Flag to enable or disable the feedback channel for upstream
communication.
* When true, components can send unanchored tuples back to their source
tasks.
*
+ * <p><b>Connection overhead:</b> when enabled, each worker hosting a user
bolt establishes
+ * an outbound Netty client connection to every remote worker hosting a
task of that bolt's
+ * immediate upstream components, regardless of stream grouping.
Connections are shared
+ * across tasks targeting the same worker, and existing outbound
connections are reused;
+ * same-worker feedback requires no network connection.</p>
+ *
* <p><b>Security:</b> feedback tuples carry a routing control signal
(e.g. per-task EWMA jitter
* used by grouping decisions), so a peer that can inject messages on the
worker transport could
* forge feedback and deterministically steer a topology's traffic to a
chosen task. This stays
diff --git a/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java
b/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java
index 0776e1eea..0d82e99f6 100644
--- a/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java
+++ b/storm-client/src/jvm/org/apache/storm/daemon/StormCommon.java
@@ -44,6 +44,7 @@ import org.apache.storm.security.auth.IAuthorizer;
import org.apache.storm.serialization.SerializationFactory;
import org.apache.storm.shade.net.minidev.json.JSONValue;
import org.apache.storm.shade.org.apache.commons.lang3.StringUtils;
+import org.apache.storm.task.GeneralTopologyContext;
import org.apache.storm.task.IBolt;
import org.apache.storm.task.WorkerTopologyContext;
import org.apache.storm.tuple.Fields;
@@ -374,6 +375,24 @@ public class StormCommon {
return Arrays.asList("task-info", "feedback");
}
+ /**
+ * Returns the distinct upstream source component ids eligible to receive
feedback, excluding system components.
+ * System components do not send upstream feedback and therefore have no
eligible sources.
+ */
+ public static Set<String>
upstreamFeedbackSourceComponents(GeneralTopologyContext context, String
componentId) {
+ Set<String> components = new HashSet<>();
+ if (Utils.isSystemId(componentId)) {
+ return components;
+ }
+ for (GlobalStreamId source : context.getSources(componentId).keySet())
{
+ String sourceComponentId = source.get_componentId();
+ if (!Utils.isSystemId(sourceComponentId)) {
+ components.add(sourceComponentId);
+ }
+ }
+ return components;
+ }
+
@SuppressWarnings("unchecked")
public static Map<String, Bolt> metricsConsumerBoltSpecs(Map<String,
Object> conf, StormTopology topology) {
Map<String, Bolt> metricsConsumerBolts = new HashMap<>();
diff --git
a/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
b/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
index 59aceb8d6..82e234fc0 100644
--- a/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
+++ b/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java
@@ -68,8 +68,10 @@ import org.apache.storm.policy.IWaitStrategy;
import org.apache.storm.security.auth.IAutoCredentials;
import org.apache.storm.serialization.ITupleSerializer;
import org.apache.storm.serialization.KryoTupleSerializer;
+import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.shade.com.google.common.collect.ImmutableMap;
import org.apache.storm.shade.com.google.common.collect.Sets;
+import org.apache.storm.task.GeneralTopologyContext;
import org.apache.storm.task.WorkerTopologyContext;
import org.apache.storm.task.WorkerUserContext;
import org.apache.storm.tuple.AddressedTuple;
@@ -800,12 +802,32 @@ public class WorkerState {
* @return seq of task ids that receive messages from this worker
*/
private Set<Integer> workerOutboundTasks() {
- WorkerTopologyContext context = getWorkerTopologyContext();
+ return computeOutboundTasks(getWorkerTopologyContext(), localTaskIds,
taskToComponent,
+ ConfigUtils.upstreamFeedbackEnable(topologyConf));
+ }
+
+ /**
+ * Computes the set of task ids this worker sends messages to.
+ *
+ * <p>Normally these are the tasks of the components that are downstream
targets of the
+ * locally-hosted components. When upstream feedback is enabled, the
source (upstream)
+ * components of the locally-hosted components are also included: feedback
tuples are routed
+ * directly to those upstream tasks (bypassing stream grouping), so they
must be reachable
+ * remote destinations and have connections established. Otherwise the
feedback is
+ * dropped and {@code TransferDrainer} logs "No remote destination
available for task".
+ */
+ @VisibleForTesting
+ static Set<Integer> computeOutboundTasks(GeneralTopologyContext context,
Collection<Integer> localTaskIds,
+ Map<Integer, String>
taskToComponent, boolean feedbackEnabled) {
Set<String> components = new HashSet<>();
for (Integer taskId : localTaskIds) {
- for (Map<String, Grouping> value :
context.getTargets(context.getComponentId(taskId)).values()) {
+ String componentId = context.getComponentId(taskId);
+ for (Map<String, Grouping> value :
context.getTargets(componentId).values()) {
components.addAll(value.keySet());
}
+ if (feedbackEnabled) {
+
components.addAll(StormCommon.upstreamFeedbackSourceComponents(context,
componentId));
+ }
}
Set<Integer> outboundTasks = new HashSet<>();
diff --git a/storm-client/src/jvm/org/apache/storm/executor/Executor.java
b/storm-client/src/jvm/org/apache/storm/executor/Executor.java
index 9a319bf07..e40d19cb8 100644
--- a/storm-client/src/jvm/org/apache/storm/executor/Executor.java
+++ b/storm-client/src/jvm/org/apache/storm/executor/Executor.java
@@ -60,7 +60,6 @@ import org.apache.storm.executor.spout.SpoutExecutor;
import org.apache.storm.generated.Bolt;
import org.apache.storm.generated.Credentials;
import org.apache.storm.generated.DebugOptions;
-import org.apache.storm.generated.GlobalStreamId;
import org.apache.storm.generated.Grouping;
import org.apache.storm.generated.SpoutSpec;
import org.apache.storm.generated.StormTopology;
@@ -603,11 +602,7 @@ public abstract class Executor implements Callable,
JCQueue.Consumer {
*/
private List<Integer> computeUpstreamTaskIds() {
Set<Integer> taskIds = new HashSet<>();
- for (GlobalStreamId source :
workerTopologyContext.getSources(componentId).keySet()) {
- String sourceComponentId = source.get_componentId();
- if (Utils.isSystemId(sourceComponentId)) {
- continue;
- }
+ for (String sourceComponentId :
StormCommon.upstreamFeedbackSourceComponents(workerTopologyContext,
componentId)) {
taskIds.addAll(workerTopologyContext.getComponentTasks(sourceComponentId));
}
return new ArrayList<>(taskIds);
diff --git a/storm-client/test/jvm/org/apache/storm/daemon/StormCommonTest.java
b/storm-client/test/jvm/org/apache/storm/daemon/StormCommonTest.java
new file mode 100644
index 000000000..a0746a509
--- /dev/null
+++ b/storm-client/test/jvm/org/apache/storm/daemon/StormCommonTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.storm.daemon;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.storm.Constants;
+import org.apache.storm.generated.GlobalStreamId;
+import org.apache.storm.generated.Grouping;
+import org.apache.storm.task.GeneralTopologyContext;
+import org.junit.jupiter.api.Test;
+
+public class StormCommonTest {
+ @Test
+ public void
testUpstreamFeedbackSourceComponentsFiltersSystemSourcesAndDeduplicatesComponents()
{
+ GeneralTopologyContext context = mock(GeneralTopologyContext.class);
+ when(context.getSources("sink")).thenReturn(Map.of(
+ new GlobalStreamId("source", "first"), new Grouping(),
+ new GlobalStreamId("source", "second"), new Grouping(),
+ new GlobalStreamId("other-source",
StormCommon.SYSTEM_STREAM_ID), new Grouping(),
+ new GlobalStreamId(Acker.ACKER_COMPONENT_ID, "ack"), new
Grouping(),
+ new GlobalStreamId(Constants.SYSTEM_COMPONENT_ID,
StormCommon.SYSTEM_STREAM_ID), new Grouping()));
+
+ assertEquals(Set.of("source", "other-source"),
StormCommon.upstreamFeedbackSourceComponents(context, "sink"));
+ }
+
+ @Test
+ public void testUpstreamFeedbackSourceComponentsForSystemComponent() {
+ GeneralTopologyContext context = mock(GeneralTopologyContext.class);
+ when(context.getSources(Acker.ACKER_COMPONENT_ID)).thenReturn(Map.of(
+ new GlobalStreamId("source", "default"), new Grouping()));
+
+ assertEquals(Set.of(),
StormCommon.upstreamFeedbackSourceComponents(context,
Acker.ACKER_COMPONENT_ID));
+ }
+
+ @Test
+ public void testUpstreamFeedbackSourceComponentsWithoutSources() {
+ GeneralTopologyContext context = mock(GeneralTopologyContext.class);
+ when(context.getSources("spout")).thenReturn(Map.of());
+
+ assertEquals(Set.of(),
StormCommon.upstreamFeedbackSourceComponents(context, "spout"));
+ }
+
+ @Test
+ public void testUpstreamFeedbackSourceComponentsWithOnlySystemSources() {
+ GeneralTopologyContext context = mock(GeneralTopologyContext.class);
+ when(context.getSources("sink")).thenReturn(Map.of(
+ new GlobalStreamId(Acker.ACKER_COMPONENT_ID, "ack"), new
Grouping()));
+
+ assertEquals(Set.of(),
StormCommon.upstreamFeedbackSourceComponents(context, "sink"));
+ }
+}
diff --git
a/storm-client/test/jvm/org/apache/storm/daemon/worker/TestUtilsForWorkerState.java
b/storm-client/test/jvm/org/apache/storm/daemon/worker/TestUtilsForWorkerState.java
index 3941f1778..145c28070 100644
---
a/storm-client/test/jvm/org/apache/storm/daemon/worker/TestUtilsForWorkerState.java
+++
b/storm-client/test/jvm/org/apache/storm/daemon/worker/TestUtilsForWorkerState.java
@@ -46,20 +46,6 @@ public class TestUtilsForWorkerState {
public static final String RESOURCE_VALUE = "resource-value";
public static WorkerState getWorkerState(Map<String, Object> conf, String
topologyId) throws IOException, TException {
- IContext context = mock(IContext.class);
-
- String assignmentId = null;
- Assignment assignment = mock(Assignment.class);
- when(assignment.get_executor_node_port()).thenReturn(new HashMap<>());
- Supervisor.Iface supervisorIface = mock(Supervisor.Iface.class);
-
when(supervisorIface.getLocalAssignmentForStorm(anyString())).thenReturn(assignment);
- SupervisorIfaceFactory supervisorIfaceFactory =
mock(SupervisorIfaceFactory.class);
- when(supervisorIfaceFactory.getIface()).thenReturn(supervisorIface);
- Supplier<SupervisorIfaceFactory> supervisorIfaceSupplier = () ->
supervisorIfaceFactory;
-
- int port = 0;
- String workerId = null;
-
Map<String, Object> topologyConf = new HashMap<>();
topologyConf.put(Config.STORM_MESSAGING_TRANSPORT,
"org.apache.storm.messaging.netty.Context");
topologyConf.put(Config.STORM_MESSAGING_NETTY_CLIENT_WORKER_THREADS,
1);
@@ -78,6 +64,23 @@ public class TestUtilsForWorkerState {
topologyConf.put(Config.TOPOLOGY_TRANSFER_BUFFER_SIZE, 1000);
topologyConf.put(Config.TOPOLOGY_TRANSFER_BATCH_SIZE, 1);
+ Assignment assignment = mock(Assignment.class);
+ when(assignment.get_executor_node_port()).thenReturn(new HashMap<>());
+ return getWorkerState(conf, topologyId, topologyConf, null,
assignment);
+ }
+
+ public static WorkerState getWorkerState(Map<String, Object> conf, String
topologyId,
+ Map<String, Object> topologyConf,
String assignmentId, Assignment assignment)
+ throws IOException, TException {
+ IContext context = mock(IContext.class);
+ Supervisor.Iface supervisorIface = mock(Supervisor.Iface.class);
+
when(supervisorIface.getLocalAssignmentForStorm(anyString())).thenReturn(assignment);
+ SupervisorIfaceFactory supervisorIfaceFactory =
mock(SupervisorIfaceFactory.class);
+ when(supervisorIfaceFactory.getIface()).thenReturn(supervisorIface);
+ Supplier<SupervisorIfaceFactory> supervisorIfaceSupplier = () ->
supervisorIfaceFactory;
+ int port = 0;
+ String workerId = null;
+
IStateStorage stateStorage = null;
IStormClusterState stormClusterState = null;
Collection<IAutoCredentials> autoCreds = null;
diff --git
a/storm-client/test/jvm/org/apache/storm/daemon/worker/WorkerOutboundTasksTest.java
b/storm-client/test/jvm/org/apache/storm/daemon/worker/WorkerOutboundTasksTest.java
new file mode 100644
index 000000000..2a66996cb
--- /dev/null
+++
b/storm-client/test/jvm/org/apache/storm/daemon/worker/WorkerOutboundTasksTest.java
@@ -0,0 +1,254 @@
+/*
+ * 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.storm.daemon.worker;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.storm.Config;
+import org.apache.storm.Constants;
+import org.apache.storm.daemon.Acker;
+import org.apache.storm.daemon.StormCommon;
+import org.apache.storm.daemon.supervisor.AdvancedFSOps;
+import org.apache.storm.generated.Assignment;
+import org.apache.storm.generated.InvalidTopologyException;
+import org.apache.storm.generated.NodeInfo;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.GeneralTopologyContext;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.testing.TestWordCounter;
+import org.apache.storm.testing.TestWordSpout;
+import org.apache.storm.thrift.TException;
+import org.apache.storm.topology.IRichBolt;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.topology.base.BaseRichBolt;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Tuple;
+import org.apache.storm.utils.ConfigUtils;
+import org.apache.storm.utils.Utils;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for worker outbound routing and topology-level feedback
configuration. The key behaviour under test:
+ * when upstream feedback is enabled, the upstream (source) component
+ * tasks of locally-hosted components must be included, because feedback
tuples are routed directly to them. Otherwise
+ * they are missing from {@code cachedTaskToNodePort}, the feedback is dropped
and {@code TransferDrainer} logs the
+ * spurious "No remote destination available for task" warning.
+ */
+public class WorkerOutboundTasksTest {
+
+ // Linear topology: spout -> bolt1 -> bolt2. Task ids are topology-global
and unique per component.
+ private static final List<Integer> SPOUT_TASKS = Arrays.asList(1, 2);
+ private static final List<Integer> BOLT1_TASKS = Arrays.asList(3, 4);
+ private static final List<Integer> BOLT2_TASKS = Arrays.asList(5, 6);
+ private static final List<Integer> ACKER_TASKS = List.of(7);
+
+ private GeneralTopologyContext context() throws InvalidTopologyException {
+ TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("spout", new NoopSpout());
+ builder.setBolt("bolt1", new NoopBolt()).shuffleGrouping("spout");
+ builder.setBolt("bolt2", new NoopBolt()).shuffleGrouping("bolt1");
+ Map<String, Object> topoConf = Utils.readStormConfig();
+ topoConf.put(Config.TOPOLOGY_WORKERS, 1);
+ topoConf.put(Config.TOPOLOGY_ACKER_EXECUTORS, ACKER_TASKS.size());
+ StormTopology topology = StormCommon.systemTopology(topoConf,
builder.createTopology());
+
+ Map<Integer, String> taskToComponent = new HashMap<>();
+ SPOUT_TASKS.forEach(t -> taskToComponent.put(t, "spout"));
+ BOLT1_TASKS.forEach(t -> taskToComponent.put(t, "bolt1"));
+ BOLT2_TASKS.forEach(t -> taskToComponent.put(t, "bolt2"));
+ ACKER_TASKS.forEach(t -> taskToComponent.put(t,
Acker.ACKER_COMPONENT_ID));
+
+ Map<String, List<Integer>> componentToSortedTasks = new HashMap<>();
+ componentToSortedTasks.put("spout", SPOUT_TASKS);
+ componentToSortedTasks.put("bolt1", BOLT1_TASKS);
+ componentToSortedTasks.put("bolt2", BOLT2_TASKS);
+ componentToSortedTasks.put(Acker.ACKER_COMPONENT_ID, ACKER_TASKS);
+
+ GeneralTopologyContext context = new GeneralTopologyContext(topology,
topoConf, taskToComponent, componentToSortedTasks,
+ new HashMap<>(), "topology-id");
+
assertTrue(context.getComponentIds().contains(Acker.ACKER_COMPONENT_ID));
+ return context;
+ }
+
+ @Test
+ public void testFeedbackDisabledOnlyDownstreamTargets() throws
InvalidTopologyException {
+ // Worker hosts a bolt1 task. Its only downstream target is bolt2, so
the upstream spout tasks are not routable.
+ GeneralTopologyContext context = context();
+ Set<Integer> outbound = WorkerState.computeOutboundTasks(context,
BOLT1_TASKS.subList(0, 1),
+ context.getTaskToComponent(), false);
+
+ BOLT2_TASKS.forEach(t -> assertTrue(outbound.contains(t), "downstream
bolt2 task " + t + " must remain routable"));
+ ACKER_TASKS.forEach(t -> assertTrue(outbound.contains(t), "acker task
" + t + " must remain routable"));
+ SPOUT_TASKS.forEach(t -> assertFalse(outbound.contains(t), "spout task
" + t + " must not be routable without feedback"));
+ }
+
+ @Test
+ public void testFeedbackEnabledIncludesUpstreamSourceTasks() throws
InvalidTopologyException {
+ // Worker hosts a bolt1 task. With feedback on, bolt1's upstream spout
tasks must also be routable so feedback
+ // tuples reach them instead of being dropped.
+ GeneralTopologyContext context = context();
+ Set<Integer> outbound = WorkerState.computeOutboundTasks(context,
BOLT1_TASKS.subList(0, 1),
+ context.getTaskToComponent(), true);
+
+ BOLT2_TASKS.forEach(t -> assertTrue(outbound.contains(t), "downstream
bolt2 task " + t + " must remain routable"));
+ ACKER_TASKS.forEach(t -> assertTrue(outbound.contains(t), "acker task
" + t + " must remain routable"));
+ SPOUT_TASKS.forEach(t -> assertTrue(outbound.contains(t), "upstream
spout task " + t + " must be routable with feedback"));
+ }
+
+ @Test
+ public void
testFeedbackDisabledColocatedSpoutMasksMissingUpstreamRouting() throws
InvalidTopologyException {
+ GeneralTopologyContext context = context();
+ List<Integer> localTasks = Arrays.asList(SPOUT_TASKS.getFirst(),
BOLT2_TASKS.getFirst());
+ Set<Integer> withoutFeedback =
WorkerState.computeOutboundTasks(context, localTasks,
+ context.getTaskToComponent(), false);
+ Set<Integer> withFeedback = WorkerState.computeOutboundTasks(context,
localTasks,
+ context.getTaskToComponent(), true);
+
+ // The spout already makes bolt1 routable for bolt2 feedback.
+ // This demonstrates how the issue with feedback delivery setup could
be hidden in particular configurations.
+ BOLT1_TASKS.forEach(t -> assertTrue(withoutFeedback.contains(t),
+ "bolt1 task " + t + " must already be routable as the
colocated spout's downstream target"));
+ BOLT1_TASKS.forEach(t -> assertTrue(withFeedback.contains(t),
+ "upstream bolt1 task " + t + " must remain routable with
feedback"));
+ ACKER_TASKS.forEach(t -> assertTrue(withoutFeedback.contains(t),
"acker task " + t + " must remain routable"));
+ ACKER_TASKS.forEach(t -> assertTrue(withFeedback.contains(t), "acker
task " + t + " must remain routable with feedback"));
+ }
+
+ @Test
+ public void
testFeedbackEnabledSystemComponentsUpstreamNotAddedAsFeedbackTasks() throws
InvalidTopologyException {
+ GeneralTopologyContext context = context();
+ Set<Integer> outbound = WorkerState.computeOutboundTasks(context,
ACKER_TASKS,
+ context.getTaskToComponent(), true);
+
+ SPOUT_TASKS.forEach(t -> assertTrue(outbound.contains(t), "spout task
" + t + " must remain routable for ack/fail messages"));
+ BOLT1_TASKS.forEach(t -> assertFalse(outbound.contains(t), "bolt1 task
" + t + " must not be routable"));
+ BOLT2_TASKS.forEach(t -> assertFalse(outbound.contains(t), "bolt2 task
" + t + " must not be routable"));
+ ACKER_TASKS.forEach(t -> assertFalse(outbound.contains(t), "acker task
" + t + " must not be routable from itself"));
+ }
+
+ @Test
+ public void testTopologyConfigEnabledFeedback() throws IOException,
TException {
+ assertOutboundTasksUseTopologyFeedbackConfiguration(true, 1);
+ }
+
+ @Test
+ public void testTopologyConfigDisabledFeedback() throws IOException,
TException {
+ assertOutboundTasksUseTopologyFeedbackConfiguration(false, 1);
+ }
+
+ @Test
+ public void
testSinkOnlyWorkerHasRemoteOutboundTasksWithFeedbackAndNoAckers() throws
IOException, TException {
+ assertOutboundTasksUseTopologyFeedbackConfiguration(true, 0);
+ }
+
+ @Test
+ public void
testSinkOnlyWorkerHasNoRemoteOutboundTasksWhenNoFeedbackAndNoAckers() throws
IOException, TException {
+ assertOutboundTasksUseTopologyFeedbackConfiguration(false, 0);
+ }
+
+ private void assertOutboundTasksUseTopologyFeedbackConfiguration(boolean
topologyFeedbackEnabled, int ackerExecutors)
+ throws IOException, TException {
+ Map<String, Object> conf = Utils.readStormConfig();
+ // set opposite to topology config to verify topology config is used
+ conf.put(Config.TOPOLOGY_UPSTREAM_FEEDBACK_ENABLE,
!topologyFeedbackEnabled);
+
+ Map<String, Object> topologyConf = new HashMap<>(conf);
+ topologyConf.put(Config.TOPOLOGY_UPSTREAM_FEEDBACK_ENABLE,
topologyFeedbackEnabled);
+ topologyConf.put(Config.TOPOLOGY_ACKER_EXECUTORS, ackerExecutors);
+ topologyConf.put(Config.TOPOLOGY_WORKERS, 1);
+
+ TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("spout", new TestWordSpout(true)).setNumTasks(1);
+ builder.setBolt("bolt1", new
TestWordCounter()).setNumTasks(1).shuffleGrouping("spout");
+ StormTopology topology = builder.createTopology();
+ Map<String, List<Integer>> componentTasks =
Utils.reverseMap(StormCommon.stormTaskInfo(topology, topologyConf));
+ int localTask = componentTasks.get("bolt1").getFirst();
+ String assignmentId = "supervisor";
+ Assignment assignment = new Assignment();
+ assignment.set_executor_node_port(Collections.singletonMap(
+ Arrays.asList((long) localTask, (long) localTask),
+ new NodeInfo(assignmentId, Collections.singleton(0L))));
+
+ ConfigUtils mockedConfigUtils = mock(ConfigUtils.class);
+ ConfigUtils previousConfigUtils =
ConfigUtils.setInstance(mockedConfigUtils);
+ try {
+ String topologyId = "feedback-config";
+ when(mockedConfigUtils.readSupervisorTopologyImpl(eq(conf),
eq(topologyId), any(AdvancedFSOps.class)))
+ .thenReturn(topology);
+ WorkerState workerState = TestUtilsForWorkerState.getWorkerState(
+ conf, topologyId, topologyConf, assignmentId, assignment);
+ try {
+ assertEquals(Set.of((int) Constants.SYSTEM_TASK_ID,
localTask), Set.copyOf(workerState.getLocalTaskIds()),
+ "worker must host only the sink bolt and the system
task");
+ assertEquals(topologyFeedbackEnabled || ackerExecutors > 0,
workerState.hasRemoteOutboundTasks(),
+ "sink-only worker must have remote outbound tasks only
if feedback or acking is enabled");
+ Set<Integer> outbound = workerState.getOutboundTasks();
+ componentTasks.get("spout").forEach(t ->
+ assertEquals(topologyFeedbackEnabled,
outbound.contains(t),
+ "upstream routing to spout task " + t + " must
follow topology configuration"));
+ } finally {
+ workerState.closeResources();
+ }
+ } finally {
+ ConfigUtils.setInstance(previousConfigUtils);
+ }
+ }
+
+ private static class NoopSpout extends BaseRichSpout {
+ @Override
+ public void open(Map<String, Object> conf, TopologyContext context,
SpoutOutputCollector collector) {
+ }
+
+ @Override
+ public void nextTuple() {
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("value"));
+ }
+ }
+
+ private static class NoopBolt extends BaseRichBolt {
+ @Override
+ public void prepare(Map<String, Object> topoConf, TopologyContext
context, OutputCollector collector) {
+ }
+
+ @Override
+ public void execute(Tuple input) {
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("value"));
+ }
+ }
+}
