reiabreu commented on code in PR #9175:
URL: https://github.com/apache/storm/pull/9175#discussion_r4303252504
##########
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));
+ }
+
+ /**
+ * 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) {
Review Comment:
This also runs for system components. A local `__acker` task has every user
component as a source, so each worker hosting an acker would connect to all of
them, although ackers never send feedback (`BoltExecutor.init` schedules the
tick only for non-system components).
```suggestion
if (feedbackEnabled && !Utils.isSystemId(componentId)) {
```
##########
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, but the executors read it
from the topology config (`Executor.java:150, 187`), so a topology that turns
feedback on in its own settings isn't fixed. `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 the helper, so which config
`workerOutboundTasks()` reads isn't tested. Reverting line 807 would still
pass. A test should build a `WorkerState` with the flag set only in the
topology conf and check `getOutboundTasks()`.
##########
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.
--
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]
