rzo1 commented on code in PR #8593:
URL: https://github.com/apache/storm/pull/8593#discussion_r3405274729
##########
storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java:
##########
@@ -131,6 +131,9 @@ public void init(ArrayList<Task> idToTask, int
idToTaskBase) throws InterruptedE
LOG.info("Prepared bolt {}:{}", componentId, taskIds);
setupTicks(false);
setupMetrics();
+ if (upstreamFeedbackEnabled) {
+ scheduleUpstreamFeedbackTick(upstreamFeedbackFreqSecs);
Review Comment:
This schedules the feedback tick for every bolt executor, including system
components. Two problems with that:
1. The acker's upstream sources are every spout and bolt in the topology, so
each acker task sends an all-VOID feedback tuple (ackers have no jitter gauges)
to every user task. Pure overhead.
2. Metrics-consumer bolts are added to the system topology after
`addUpstreamFeedback`, so they never get the `__feedback` stream declared. When
their tick fires, `Task.getTuple` creates a `TupleImpl` on an undeclared
stream. In local mode (`doSanityCheck`) the `getComponentOutputFields` lookup
throws and kills the executor.
Guarding with `!Utils.isSystemId(componentId)` here should fix both.
##########
storm-client/src/jvm/org/apache/storm/utils/ConfigUtils.java:
##########
@@ -198,6 +198,29 @@ public static boolean ewmaEnable(Map<String, Object> conf)
{
return ObjectReader.getBoolean(value, false);
}
+ public static boolean upstreamFeedbackEnable(Map<String, Object> conf) {
+ Object value = conf.get(Config.TOPOLOGY_UPSTREAM_FEEDBACK_ENABLE);
+ if (value == null) {
+ return false;
+ }
+ return ObjectReader.getBoolean(value, false);
+ }
+
+ public static String upstreamFeedbackStreamId(Map<String, Object> conf) {
+ Object value = conf.get(Config.TOPOLOGY_UPSTREAM_FEEDBACK_STREAM_ID);
+ return ObjectReader.getString(value);
Review Comment:
`ObjectReader.getString(value)` throws on null, and this is called
unconditionally in the `Executor` constructor even when the feature is
disabled, so any conf map without the key crashes every executor at startup.
The javadoc in `Config` promises a `__feedback` default, so this should be
```java
return ObjectReader.getString(value, "__feedback");
```
Though see my comment on `Config.java`, I'd prefer dropping the config
entirely.
##########
storm-client/src/jvm/org/apache/storm/Config.java:
##########
@@ -608,8 +608,45 @@ public class Config extends HashMap<String, Object> {
*
* @see <a href="https://www.rfc-editor.org/rfc/rfc1889#appendix-A.8";>RFC
1889 §A.8</a>
*/
- @CustomValidator(validatorClass =
ConfigValidation.EwmaSmoothingFactorValidator.class)
+ @CustomValidator(validatorClass =
ConfigValidation.ZeroOneOpenIntervalValidator.class)
public static final String TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR =
"topology.stats.ewma.smoothing.factor";
+ /**
+ * Flag to enable or disable the feedback channel for upstream
communication.
+ * When true, components can send unanchored tuples back to their source
tasks.
+ */
+ @IsBoolean
+ public static final String TOPOLOGY_UPSTREAM_FEEDBACK_ENABLE =
"topology.upstream.feedback.enable";
+ /**
+ * The specific stream ID used for upstream feedback communication.
+ * Defaults to "__feedback" if not explicitly configured.
+ */
+ @IsString
+ public static final String TOPOLOGY_UPSTREAM_FEEDBACK_STREAM_ID =
"topology.upstream.feedback.stream";
Review Comment:
User stream ids can't start with `__`, but this config is only `@IsString`.
If someone sets it to `"default"`, `addUpstreamFeedback` calls
`put_to_streams("default", ...)` on every component and silently replaces
user-declared output fields topology-wide. Is there a use case for making this
configurable at all? `FEEDBACK_TICK_STREAM_ID` is already a constant, I'd put
this one next to it in `Constants`. If it has to stay configurable, validate
the `__` prefix.
##########
storm-client/src/jvm/org/apache/storm/grouping/LoadAwareShuffleGrouping.java:
##########
@@ -24,6 +24,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.storm.Config;
+import org.apache.storm.executor.ChildEwmaStats;
Review Comment:
Same here, unused import. Please drop.
##########
storm-client/src/jvm/org/apache/storm/grouping/JitterAwareStreamGrouping.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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.grouping;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.storm.executor.ChildEwmaStats;
+import org.apache.storm.generated.GlobalStreamId;
+import org.apache.storm.task.WorkerTopologyContext;
+
+/**
+ * A {@link CustomStreamGrouping} that routes each tuple to the downstream
(child) task with the lowest
+ * jitter, as reported back to the emitting task through upstream feedback and
aggregated by
+ * {@link ChildEwmaStats}. Candidates are ordered with {@link
ChildEwmaStats#compareByJitter}, so a lower
+ * {@code __execute-jitter} wins first, then {@code __process-jitter}, then
{@code __complete-jitter}.
+ * Until a source task has any feedback data — and for targets that have not
reported yet — the grouping
+ * falls back to round-robin over the target tasks, so it degrades to an even
spread rather than pinning a
+ * single task.
+ */
+public class JitterAwareStreamGrouping implements
LoadAwareCustomStreamGrouping {
+
+ private final AtomicInteger roundRobin = new AtomicInteger();
+ private List<Integer> targetTasks;
+ private ChildEwmaStats stats;
+
+ @Override
+ public void refreshLoad(LoadMapping loadMapping) {
+ // load mapping agnostic
+ }
+
+ @Override
+ public void registerEwmaStats(ChildEwmaStats childEwmaStats) {
+ this.stats = childEwmaStats;
+ }
+
+ @Override
+ public void prepare(WorkerTopologyContext context, GlobalStreamId stream,
List<Integer> targetTasks) {
+ this.targetTasks = targetTasks;
+ }
+
+ @Override
+ public List<Integer> chooseTasks(int taskId, List<Object> values) {
Review Comment:
Winner-take-all between feedback updates means every upstream task pins the
same lowest-jitter target for the whole feedback interval, overloads it, and
the herd moves on at the next report. Consider power-of-two-choices or weighted
selection, see `LoadAwareShuffleGrouping`.
Small doc nit: the comment says "averaged value" but
`EwmaFeedbackRecord.aggregate` takes the max across per-source gauges.
##########
storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java:
##########
@@ -74,6 +74,7 @@ public static Kryo getKryo(Map<String, Object> conf) {
k.register(Values.class);
k.register(org.apache.storm.metric.api.IMetricsConsumer.DataPoint.class);
k.register(org.apache.storm.metric.api.IMetricsConsumer.TaskInfo.class);
+ k.register(org.apache.storm.executor.EwmaFeedbackRecord.class);
Review Comment:
Registering here shifts the sequential Kryo ids of everything below
(`ConsList`, `BackPressureStatus`, `NodeInfo`, ...). `BackPressureStatus`
crosses the wire between workers, and if two communicating JVMs ever disagree
on the list (mixed-version supervisors during a rolling upgrade)
deserialization corrupts silently. Please append new registrations at the end
of the list.
##########
storm-client/src/jvm/org/apache/storm/daemon/GrouperFactory.java:
##########
@@ -20,6 +20,7 @@
import java.util.Set;
import org.apache.storm.Config;
import org.apache.storm.Thrift;
+import org.apache.storm.executor.ChildEwmaStats;
Review Comment:
This file only adds an unused import, leftover from an earlier revision?
Please drop.
##########
examples/storm-perf/src/main/java/org/apache/storm/perf/JitterAwareGroupingTopology.java:
##########
@@ -0,0 +1,296 @@
+/*
+ * 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.perf;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.locks.LockSupport;
+import org.apache.storm.Config;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.grouping.JitterAwareStreamGrouping;
+import org.apache.storm.perf.spout.FileReadSpout;
+import org.apache.storm.perf.utils.Helper;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
+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.tuple.Values;
+import org.apache.storm.utils.Utils;
+
+/**
+ * Benchmark for {@link JitterAwareStreamGrouping} in a word-count pipeline
where worker tasks have
+ * artificially skewed processing latencies.
+ *
+ * <p>Pipeline: {@code GenSpout -> SplitterBolt -> JitteryWorkerBolt ->
SinkBolt}
+ *
+ * <p>{@code JitteryWorkerBolt} tasks have task-index-dependent processing
delays. Task 0 is fast;
+ * each subsequent task is progressively slower. This mimics real-world
conditions (GC pressure,
+ * I/O, resource contention) where downstream tasks diverge in responsiveness.
With upstream
+ * feedback enabled, {@link JitterAwareStreamGrouping} routes more tuples to
the fastest tasks,
+ * improving throughput and reducing complete latency by ≥10% compared to
plain round-robin.
Review Comment:
The javadoc claims a 10%+ improvement over round-robin. Let's keep
performance claims out of the code until the benchmark questions are settled.
Also, a missing `input.file` currently fails with a bare NPE from `new
FileInputStream(null)`, a short usage message would help.
--
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]
