reiabreu commented on code in PR #9175:
URL: https://github.com/apache/storm/pull/9175#discussion_r4303252492
##########
storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java:
##########
@@ -800,12 +803,37 @@ private StormTimer mkHaltingTimer(String name) {
* @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(conf));
Review Comment:
The feedback flag is read from the daemon config (`conf`), but executors
read it from the topology config (`Executor.java:150, 187`). If a topology sets
`topology.upstream.feedback.enable` in its own config, `WorkerState` doesn't
see it, and the upstream tasks on other workers are not made routable.
`topologyConf` is assigned at line 208, before this runs at line 231.
```suggestion
ConfigUtils.upstreamFeedbackEnable(topologyConf));
```
##########
storm-client/test/jvm/org/apache/storm/daemon/worker/WorkerOutboundTasksTest.java:
##########
@@ -0,0 +1,124 @@
+/*
+ * 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 java.util.Arrays;
+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.generated.StormTopology;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.GeneralTopologyContext;
+import org.apache.storm.task.TopologyContext;
+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.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link WorkerState#computeOutboundTasks}, which decides the
set of task ids a worker must be able to
+ * route messages to. 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 GeneralTopologyContext context() {
+ TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("spout", new NoopSpout());
+ builder.setBolt("bolt1", new NoopBolt()).shuffleGrouping("spout");
+ builder.setBolt("bolt2", new NoopBolt()).shuffleGrouping("bolt1");
+ StormTopology topology = 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"));
+
+ Map<String, List<Integer>> componentToSortedTasks = new HashMap<>();
+ componentToSortedTasks.put("spout", SPOUT_TASKS);
+ componentToSortedTasks.put("bolt1", BOLT1_TASKS);
+ componentToSortedTasks.put("bolt2", BOLT2_TASKS);
+
+ return new GeneralTopologyContext(topology, new Config(),
taskToComponent, componentToSortedTasks,
+ new HashMap<>(), "topology-id");
+ }
+
+ @Test
+ public void feedbackDisabled_onlyDownstreamTargets() {
+ // 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);
+
+ assertEquals(new java.util.HashSet<>(BOLT2_TASKS), outbound);
+ SPOUT_TASKS.forEach(t -> assertFalse(outbound.contains(t), "spout task
" + t + " must not be routable without feedback"));
+ }
+
+ @Test
+ public void feedbackEnabled_includesUpstreamSourceTasks() {
Review Comment:
The tests pass a literal boolean to `computeOutboundTasks`, so which config
`workerOutboundTasks()` reads isn't tested. Reverting line 807 would still
pass. A test should build a `WorkerState` with the flag set in the topology
conf but not in the daemon conf, then check `getOutboundTasks()`. How heavy it
is to construct a `WorkerState` in a unit test hasn't been checked;
`WorkerStateTest` may show a pattern.
##########
storm-client/test/jvm/org/apache/storm/daemon/worker/WorkerOutboundTasksTest.java:
##########
@@ -0,0 +1,124 @@
+/*
+ * 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 java.util.Arrays;
+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.generated.StormTopology;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.GeneralTopologyContext;
+import org.apache.storm.task.TopologyContext;
+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.Tuple;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link WorkerState#computeOutboundTasks}, which decides the
set of task ids a worker must be able to
+ * route messages to. 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 GeneralTopologyContext context() {
Review Comment:
This context has only user components, with no `__acker` or `__system`, so
the over-inclusion described on `WorkerState.java:829` can't show up. Please
add an acker and assert that a worker hosting only an acker gains no feedback
recipients. Also, the PR description lists a third test (spout and `bolt2`
co-located, so `bolt1` is already routable), but this file has only two.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
