ErnestasVaiciukeviciusAdform commented on code in PR #9175:
URL: https://github.com/apache/storm/pull/9175#discussion_r4311456443
##########
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:
System component check is present on line 832, but yes - we could cut off
that case earlier before even iterating the sources. Will adjust, thx.
--
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]
