You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@storm.apache.org by "GGraziadei (via GitHub)" <gi...@apache.org> on 2026/05/05 19:34:48 UTC
[PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
GGraziadei opened a new pull request, #8586:
URL: https://github.com/apache/storm/pull/8586
## What is the purpose of the change
In deterministic real-time processing, predictability of latency is as important as latency itself. This is a constraint to building a deterministic system.
- Mcro-burst detection: high jitter reveals short spikes that average latency smooths out.
- Compliance: modern SLAs rely on percentiles (e.g., P99). Jitter is a strong leading indicator of tail-latency degradation.
- Root Cause Analysis: high component jitter means GC pressure or resource contention; instead, high global jitter with stable components suggests network congestion or shuffle bottlenecks.
- Bottleneck identification: jitter enables precise identification of where bottlenecks occur in the topology and helps distinguish their underlying causes, making performance issues easier to diagnose and resolve.
To ensure negligible performance impact, I propose to use an Exponentially Weighted Moving Average (EWMA), following RFC 1889 logic https://www.rfc-editor.org/rfc/rfc1889#appendix-A.8
Mathematical Model:
J_new = J_old + (|D_current - D_previous| - J_old) * smoothing_factor
Performance impact
- Minimal computational overhead: by utilizing an EWMA.
- Memory efficiency: only two persistent variables (8 bytes) per task.
- System calls: no system calls required to track the latency (the latencies are already computed).
## How was the change tested
- Unit test: introduced new test cases for `Config`, `TaskMetrics`, `EwmaGauge`
- Smoke test in local: registered a topology metrics reporter and persisted captured metrics in the attached file
- The package `metrics2` doesn't affect it.
[worker_log.zip](https://github.com/user-attachments/files/27413715/worker_log.zip)
Example results in worker logs
```
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] storm.worker.WordCountTopology-4-1777995769.ggraziadei-ThinkPad-E14-Gen-5.count.default.10.6700-__emit-count-default.m1_rate
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] value = 30.0
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] storm.worker.WordCountTopology-4-1777995769.ggraziadei-ThinkPad-E14-Gen-5.count.default.10.6700-__execute-count-split:default.m1_rate
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] value = 30.0
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] storm.worker.WordCountTopology-4-1777995769.ggraziadei-ThinkPad-E14-Gen-5.count.default.10.6700-__execute-latency-split:default
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] value = 0.0
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] storm.worker.WordCountTopology-4-1777995769.ggraziadei-ThinkPad-E14-Gen-5.count.default.10.6700-__execute-rfc1889a-jitter-split:default
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] value = 0.2557194505051832
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] storm.worker.WordCountTopology-4-1777995769.ggraziadei-ThinkPad-E14-Gen-5.count.default.10.6700-__process-latency-split:default
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] value = 0.3333333333333333
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] storm.worker.WordCountTopology-4-1777995769.ggraziadei-ThinkPad-E14-Gen-5.count.default.10.6700-__process-rfc1889a-jitter-split:default
2026-05-05 17:52:07.993 c.c.m.ConsoleReporter metrics-console-reporter-1-thread-1 [INFO] value = 0.145830156234796
```
In the context of #8583
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #8586:
URL: https://github.com/apache/storm/pull/8586#discussion_r3209327950
##########
storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.metrics2;
+
+import static org.apache.storm.utils.ConfigUtils.RFC1889_ALPHA;
+
+import com.codahale.metrics.Gauge;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Lock-free jitter estimator following RFC 1889 Section 6.3.1.
+ * The jitter accumulator is stored as raw IEEE 754 bits in an AtomicLong
+ * so that CAS can be used without locks.
+ * Thread safety: addValue is lock-free; getValue is wait-free.
+ */
+public class EwmaGauge implements Gauge<Double> {
+
+ private static final long UNSEEDED = Long.MIN_VALUE;
+ private static final long ZERO_BITS = Double.doubleToLongBits(0.0);
+
+ private final AtomicLong lastTransit = new AtomicLong(UNSEEDED);
+ private final AtomicLong jitterBits = new AtomicLong(ZERO_BITS);
+ private final double alpha;
+
+ EwmaGauge(double alpha) {
+ if (alpha <= 0.0 || alpha >= 1.0 || Double.isNaN(alpha)) {
+ throw new IllegalArgumentException(
+ "alpha must be in (0, 1), got: " + alpha);
+ }
+ this.alpha = alpha;
+ }
+
+ EwmaGauge() {
+ this(RFC1889_ALPHA); // 1.0 / 16.0
+ }
+
+ /**
+ * Update the jitter estimate.
+ *
+ * @param transitMs transit time for this tuple: {@code arrival - timestamp}
+ * Negative values are silently ignored.
+ */
+ public void addValue(long transitMs) {
+ if (transitMs < 0) {
+ return;
+ }
+
+ // Seed on the very first packet: store transit, nothing to diff against yet.
+ if (lastTransit.compareAndSet(UNSEEDED, transitMs)) {
+ return;
+ }
+
+ long prev = lastTransit.getAndSet(transitMs);
+ if (prev == UNSEEDED) {
+ // Lost a race during seeding; prev is not a real transit value.
+ return;
+ }
Review Comment:
You are right! UNSEEDED is never restored once set, so this branch is indeed unreachable. I'll remove it.
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #8586:
URL: https://github.com/apache/storm/pull/8586#discussion_r3209524840
##########
storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java:
##########
@@ -849,6 +849,23 @@ public void validateField(String name, Object o) {
}
}
+ public static class EwmaSmoothingFactorValidator extends Validator {
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ return;
+ }
+ if (o instanceof Number) {
+ double alpha = ((Number) o).doubleValue();
+ if (alpha > 0.0 && alpha < 1.0) {
+ return;
+ }
+ }
+ throw new IllegalArgumentException(
+ "Field " + name + " must be a number in the open interval (0, 1), got: " + o);
Review Comment:
Thank you, fixed and added a new test case.
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on PR #8586:
URL: https://github.com/apache/storm/pull/8586#issuecomment-4386170632
Hello @rzo1 thank you for your comment.
You have reason, there is an implementation error, and I am fixing it.
If I do not update the jitter when the latency is stable `d==0`, the jitter doesn't decrease, and this is weird.
Regarding the test case `Zero deviation decays jitter toward zero` (alpha=0.5), the correct status serialization is reported here:
- lastLat = UNSEED; lat=0; j=0
- lastLat=0; lat=10; d=5; j=0+alpha* (d-j)=2.5
- lastLat=10; lat=10; d=0; j=2.5 + alpha * (d-j) = 2.5 - alpha * 2.5 = 1.25
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #8586:
URL: https://github.com/apache/storm/pull/8586#discussion_r3209425159
##########
storm-client/src/jvm/org/apache/storm/metrics2/TaskMetrics.java:
##########
@@ -27,19 +29,26 @@ public class TaskMetrics {
private static final String METRIC_NAME_TRANSFERRED = "__transfer-count";
private static final String METRIC_NAME_EXECUTED = "__execute-count";
private static final String METRIC_NAME_PROCESS_LATENCY = "__process-latency";
+ private static final String METRIC_NAME_PROCESS_RFC_1889a_JITTER = "__process-rfc1889a-jitter";
private static final String METRIC_NAME_COMPLETE_LATENCY = "__complete-latency";
+ private static final String METRIC_NAME_COMPLETE_RFC_1889a_JITTER = "__complete-rfc1889a-jitter";
private static final String METRIC_NAME_EXECUTE_LATENCY = "__execute-latency";
+ private static final String METRIC_NAME_EXECUTE_RFC_1889a_JITTER = "__execute-rfc1889a-jitter";
Review Comment:
Thank you! Changed in `__complete-jitter`
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "rzo1 (via GitHub)" <gi...@apache.org>.
rzo1 commented on code in PR #8586:
URL: https://github.com/apache/storm/pull/8586#discussion_r3201137167
##########
conf/defaults.yaml:
##########
@@ -278,6 +278,7 @@ topology.max.task.parallelism: null
topology.max.spout.pending: null # ideally should be larger than topology.producer.batch.size. (esp. if topology.batch.flush.interval.millis=0)
topology.state.synchronization.timeout.secs: 60
topology.stats.sample.rate: 0.05
+topology.stats.ewma.enable: false
Review Comment:
`topology.stats.ewma.smoothing_factor` isn't represented here even though it's a tunable. Either add `topology.stats.ewma.smoothing.factor: 0.0625` (or `null` with a comment about the `RFC1889_ALPHA` fallback) so operators can discover the knob via `defaults.yaml`, or call out the default in the `Config.java` Javadoc — right now the `1/16` default is invisible outside the source.
##########
storm-client/src/jvm/org/apache/storm/Config.java:
##########
@@ -596,6 +596,20 @@ public class Config extends HashMap<String, Object> {
*/
@IsPositiveNumber
public static final String TOPOLOGY_STATS_SAMPLE_RATE = "topology.stats.sample.rate";
+ /**
+ * Enabling jitter streaming calculation (RFC 1889a).
+ *
+ * @see <a href="https://www.rfc-editor.org/rfc/rfc1889#appendix-A.8">RFC 1889 Appendix A.8</a>
+ */
+ @IsBoolean
+ public static final String TOPOLOGY_STATS_EWMA_ENABLE = "topology.stats.ewma.enable";
+ /**
+ * The smoothing factor (alpha) used for exponential jitter calculation (RFC 1889a).
+ *
+ * @see <a href="https://www.rfc-editor.org/rfc/rfc1889#appendix-A.8">RFC 1889 Appendix A.8</a>
+ */
+ @CustomValidator(validatorClass = ConfigValidation.EwmaSmoothingFactorValidator.class)
+ public static final String TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR = "topology.stats.ewma.smoothing_factor";
Review Comment:
Underscore breaks Storm's `dot.case` config convention. Every neighbour key uses dots (`topology.stats.sample.rate`, `topology.builtin.metrics.bucket.size.secs`, …). Suggest `topology.stats.ewma.smoothing.factor`. Worth changing now — once released it's a public surface.
##########
storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.metrics2;
+
+import static org.apache.storm.utils.ConfigUtils.RFC1889_ALPHA;
+
+import com.codahale.metrics.Gauge;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Lock-free jitter estimator following RFC 1889 Section 6.3.1.
Review Comment:
RFC 1889 §6.3.1 is the *Receiver Reports* section, not the jitter algorithm. The algorithm is in §A.8. Suggest: `Lock-free jitter estimator following RFC 1889 §A.8 / RFC 3550 §A.8.`
##########
storm-client/src/jvm/org/apache/storm/metrics2/TaskMetrics.java:
##########
@@ -27,19 +29,26 @@ public class TaskMetrics {
private static final String METRIC_NAME_TRANSFERRED = "__transfer-count";
private static final String METRIC_NAME_EXECUTED = "__execute-count";
private static final String METRIC_NAME_PROCESS_LATENCY = "__process-latency";
+ private static final String METRIC_NAME_PROCESS_RFC_1889a_JITTER = "__process-rfc1889a-jitter";
private static final String METRIC_NAME_COMPLETE_LATENCY = "__complete-latency";
+ private static final String METRIC_NAME_COMPLETE_RFC_1889a_JITTER = "__complete-rfc1889a-jitter";
private static final String METRIC_NAME_EXECUTE_LATENCY = "__execute-latency";
+ private static final String METRIC_NAME_EXECUTE_RFC_1889a_JITTER = "__execute-rfc1889a-jitter";
Review Comment:
Two issues with these metric names:
1. **`rfc1889a` suffix isn't a real RFC label** — the algorithm is in *Appendix A.8* of RFC 1889. The `a` reads like a version suffix. Suggest `__complete-rfc1889-jitter` etc., or simply `__complete-jitter` (the RFC reference belongs in `Metrics.md`, not in every metric series name).
2. **Inconsistent identifier casing** — the constants spell it `RFC_1889a` (camel-case mid-word with lowercase `a`). Convention in this file is all-uppercase: `RFC_1889_JITTER` or `RFC1889_JITTER`.
These metric names become public API the moment the PR ships — much easier to fix now than after dashboards are wired up.
##########
storm-client/src/jvm/org/apache/storm/metrics2/TaskMetrics.java:
##########
@@ -27,19 +29,26 @@ public class TaskMetrics {
private static final String METRIC_NAME_TRANSFERRED = "__transfer-count";
private static final String METRIC_NAME_EXECUTED = "__execute-count";
private static final String METRIC_NAME_PROCESS_LATENCY = "__process-latency";
+ private static final String METRIC_NAME_PROCESS_RFC_1889a_JITTER = "__process-rfc1889a-jitter";
private static final String METRIC_NAME_COMPLETE_LATENCY = "__complete-latency";
+ private static final String METRIC_NAME_COMPLETE_RFC_1889a_JITTER = "__complete-rfc1889a-jitter";
private static final String METRIC_NAME_EXECUTE_LATENCY = "__execute-latency";
+ private static final String METRIC_NAME_EXECUTE_RFC_1889a_JITTER = "__execute-rfc1889a-jitter";
private static final String METRIC_NAME_CAPACITY = "__capacity";
private final ConcurrentMap<String, RateCounter> rateCounters = new ConcurrentHashMap<>();
- private final ConcurrentMap<String, RollingAverageGauge> gauges = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, Gauge> gauges = new ConcurrentHashMap<>();
Review Comment:
Going to a raw `ConcurrentMap<String, Gauge>` loses type info that the previous `ConcurrentMap<String, RollingAverageGauge>` had. Cleaner would be `ConcurrentMap<String, Gauge<?>>` — that keeps the wildcard typing, lets `getOrCreateGauge` work unchanged, and the only raw cast moves to `registerGauge` where you already have `@SuppressWarnings({"unchecked", "rawtypes"})`.
##########
storm-client/src/jvm/org/apache/storm/metrics2/TaskMetrics.java:
##########
@@ -135,18 +166,54 @@ private RateCounter getRateCounter(String metricName, String streamId) {
}
private RollingAverageGauge getRollingAverageGauge(String metricName, String streamId) {
- RollingAverageGauge gauge = this.gauges.get(metricName);
- if (gauge == null) {
+ return getOrCreateGauge(metricName, streamId, RollingAverageGauge.class, this.rollingAverageGaugeFactory);
+ }
+
+ private EwmaGauge getExponentialWeightedMobileAverageGauge(String metricName, String streamId) {
Review Comment:
Typo: "Mobile" → "Moving". EWMA = Exponentially Weighted **Moving** Average. (The field name `ewmaSmoothingFactor` is fine; just the helper method.)
##########
storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java:
##########
@@ -849,6 +849,23 @@ public void validateField(String name, Object o) {
}
}
+ public static class EwmaSmoothingFactorValidator extends Validator {
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ return;
+ }
+ if (o instanceof Number) {
+ double alpha = ((Number) o).doubleValue();
+ if (alpha > 0.0 && alpha < 1.0) {
+ return;
+ }
+ }
+ throw new IllegalArgumentException(
+ "Field " + name + " must be a number in the open interval (0, 1), got: " + o);
Review Comment:
This validator only accepts `Number`, but `ConfigUtils.ewmaSmoothingFactor` calls `ObjectReader.getDouble(value)` which also parses strings like `"0.5"`. Other numeric configs in Storm (e.g. `topology.stats.sample.rate`) accept stringified numbers from YAML, so the asymmetry is surprising — a value would be rejected at validation time but accepted at runtime if validation is bypassed (programmatic conf, tests, etc.).
Suggest mirroring what the runtime does: try `ObjectReader.getDouble(o)` inside a try/catch, then range-check. That way the validator and the runtime parser agree.
##########
storm-client/src/jvm/org/apache/storm/Config.java:
##########
@@ -596,6 +596,20 @@ public class Config extends HashMap<String, Object> {
*/
@IsPositiveNumber
public static final String TOPOLOGY_STATS_SAMPLE_RATE = "topology.stats.sample.rate";
+ /**
+ * Enabling jitter streaming calculation (RFC 1889a).
Review Comment:
Minor: "RFC 1889a" isn't a real RFC label — the algorithm lives in **Appendix A.8** of RFC 1889 (and was carried unchanged into RFC 3550 §A.8). Suggest "RFC 1889 §A.8" or "RFC 3550 §A.8" everywhere it appears in this PR (Javadoc, metric names, docs). RFC 1889 has been obsolete since 2003, so a forward link to RFC 3550 is friendlier to readers.
##########
storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.metrics2;
+
+import static org.apache.storm.utils.ConfigUtils.RFC1889_ALPHA;
+
+import com.codahale.metrics.Gauge;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Lock-free jitter estimator following RFC 1889 Section 6.3.1.
+ * The jitter accumulator is stored as raw IEEE 754 bits in an AtomicLong
+ * so that CAS can be used without locks.
+ * Thread safety: addValue is lock-free; getValue is wait-free.
+ */
+public class EwmaGauge implements Gauge<Double> {
+
+ private static final long UNSEEDED = Long.MIN_VALUE;
+ private static final long ZERO_BITS = Double.doubleToLongBits(0.0);
+
+ private final AtomicLong lastTransit = new AtomicLong(UNSEEDED);
+ private final AtomicLong jitterBits = new AtomicLong(ZERO_BITS);
+ private final double alpha;
+
+ EwmaGauge(double alpha) {
+ if (alpha <= 0.0 || alpha >= 1.0 || Double.isNaN(alpha)) {
+ throw new IllegalArgumentException(
+ "alpha must be in (0, 1), got: " + alpha);
+ }
+ this.alpha = alpha;
+ }
+
+ EwmaGauge() {
+ this(RFC1889_ALPHA); // 1.0 / 16.0
+ }
+
+ /**
+ * Update the jitter estimate.
+ *
+ * @param transitMs transit time for this tuple: {@code arrival - timestamp}
+ * Negative values are silently ignored.
+ */
+ public void addValue(long transitMs) {
+ if (transitMs < 0) {
+ return;
+ }
+
+ // Seed on the very first packet: store transit, nothing to diff against yet.
+ if (lastTransit.compareAndSet(UNSEEDED, transitMs)) {
+ return;
+ }
+
+ long prev = lastTransit.getAndSet(transitMs);
+ if (prev == UNSEEDED) {
+ // Lost a race during seeding; prev is not a real transit value.
+ return;
+ }
+
+ double d = Math.abs(transitMs - prev);
Review Comment:
Worth a one-line comment that this is safe from the well-known `Math.abs(Long.MIN_VALUE) == Long.MIN_VALUE` pathology *because* `transitMs >= 0 && prev >= 0` (enforced by the negative-guard at line 55 + the seed path), so `transitMs - prev ∈ [-Long.MAX_VALUE, Long.MAX_VALUE]`. Otherwise a future reader is going to wonder.
##########
storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.metrics2;
+
+import static org.apache.storm.utils.ConfigUtils.RFC1889_ALPHA;
+
+import com.codahale.metrics.Gauge;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Lock-free jitter estimator following RFC 1889 Section 6.3.1.
+ * The jitter accumulator is stored as raw IEEE 754 bits in an AtomicLong
+ * so that CAS can be used without locks.
+ * Thread safety: addValue is lock-free; getValue is wait-free.
+ */
+public class EwmaGauge implements Gauge<Double> {
+
+ private static final long UNSEEDED = Long.MIN_VALUE;
+ private static final long ZERO_BITS = Double.doubleToLongBits(0.0);
+
+ private final AtomicLong lastTransit = new AtomicLong(UNSEEDED);
+ private final AtomicLong jitterBits = new AtomicLong(ZERO_BITS);
+ private final double alpha;
+
+ EwmaGauge(double alpha) {
+ if (alpha <= 0.0 || alpha >= 1.0 || Double.isNaN(alpha)) {
+ throw new IllegalArgumentException(
+ "alpha must be in (0, 1), got: " + alpha);
+ }
+ this.alpha = alpha;
+ }
+
+ EwmaGauge() {
+ this(RFC1889_ALPHA); // 1.0 / 16.0
+ }
+
+ /**
+ * Update the jitter estimate.
+ *
+ * @param transitMs transit time for this tuple: {@code arrival - timestamp}
+ * Negative values are silently ignored.
+ */
+ public void addValue(long transitMs) {
+ if (transitMs < 0) {
+ return;
+ }
+
+ // Seed on the very first packet: store transit, nothing to diff against yet.
+ if (lastTransit.compareAndSet(UNSEEDED, transitMs)) {
+ return;
+ }
+
+ long prev = lastTransit.getAndSet(transitMs);
+ if (prev == UNSEEDED) {
+ // Lost a race during seeding; prev is not a real transit value.
+ return;
+ }
+
+ double d = Math.abs(transitMs - prev);
+
+ long currentBits;
+ long updatedBits;
+ do {
+ currentBits = jitterBits.get();
+ double currentJitter = Double.longBitsToDouble(currentBits);
+ double updatedJitter = currentJitter + alpha * (d - currentJitter);
+ updatedBits = Double.doubleToLongBits(updatedJitter);
+ } while (!jitterBits.compareAndSet(currentBits, updatedBits));
+ }
+
+ /**
+ * Returns the current jitter estimate in timestamp units.
+ */
+ @Override
+ public Double getValue() {
+ return Double.longBitsToDouble(jitterBits.get());
+ }
+}
Review Comment:
Missing newline at EOF. Same in `EwmaGaugeTest.java` and `TaskMetricsTest.java` — checkstyle will probably flag these.
##########
storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.metrics2;
+
+import static org.apache.storm.utils.ConfigUtils.RFC1889_ALPHA;
+
+import com.codahale.metrics.Gauge;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Lock-free jitter estimator following RFC 1889 Section 6.3.1.
+ * The jitter accumulator is stored as raw IEEE 754 bits in an AtomicLong
+ * so that CAS can be used without locks.
+ * Thread safety: addValue is lock-free; getValue is wait-free.
+ */
+public class EwmaGauge implements Gauge<Double> {
+
+ private static final long UNSEEDED = Long.MIN_VALUE;
+ private static final long ZERO_BITS = Double.doubleToLongBits(0.0);
+
+ private final AtomicLong lastTransit = new AtomicLong(UNSEEDED);
+ private final AtomicLong jitterBits = new AtomicLong(ZERO_BITS);
+ private final double alpha;
+
+ EwmaGauge(double alpha) {
+ if (alpha <= 0.0 || alpha >= 1.0 || Double.isNaN(alpha)) {
+ throw new IllegalArgumentException(
+ "alpha must be in (0, 1), got: " + alpha);
+ }
+ this.alpha = alpha;
+ }
+
+ EwmaGauge() {
+ this(RFC1889_ALPHA); // 1.0 / 16.0
+ }
+
+ /**
+ * Update the jitter estimate.
+ *
+ * @param transitMs transit time for this tuple: {@code arrival - timestamp}
+ * Negative values are silently ignored.
+ */
+ public void addValue(long transitMs) {
+ if (transitMs < 0) {
+ return;
+ }
+
+ // Seed on the very first packet: store transit, nothing to diff against yet.
+ if (lastTransit.compareAndSet(UNSEEDED, transitMs)) {
+ return;
+ }
+
+ long prev = lastTransit.getAndSet(transitMs);
+ if (prev == UNSEEDED) {
+ // Lost a race during seeding; prev is not a real transit value.
+ return;
+ }
Review Comment:
Nothing in this class ever resets `lastTransit` back to `UNSEEDED`, so once the seed CAS succeeds, `getAndSet` can't return `UNSEEDED` — this branch is unreachable. Either drop it, or keep it and add a comment that it's defensive against a future caller adding a reset path. Right now the comment ("Lost a race during seeding") is incorrect: the seed CAS is non-revertible, so a thread that lost the seed race still observes a real `transitMs` here.
##########
storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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
+ * <p>
Review Comment:
`<p>` tags in the license header are inconsistent with the rest of `storm-client` (the standard ASF header uses blank lines between paragraphs — see e.g. `TaskMetrics.java` in this same PR). Same applies to `EwmaGaugeTest.java` (`<p/>`) and `TaskMetricsTest.java` (`<p/>`). Looks like an IDE auto-formatter artifact.
##########
storm-client/test/jvm/org/apache/storm/metrics2/TaskMetricsTest.java:
##########
@@ -0,0 +1,487 @@
+/*
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.metrics2;
+
+import com.codahale.metrics.Gauge;
+import org.apache.storm.task.WorkerTopologyContext;
+import org.apache.storm.utils.ConfigUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+@ExtendWith(MockitoExtension.class)
+class TaskMetricsTest {
+
+ private static final String TOPOLOGY_ID = "test-topology-1";
+ private static final String COMPONENT_ID = "test-bolt";
+ private static final Integer TASK_ID = 42;
+ private static final Integer WORKER_PORT = 6700;
+ private static final String STREAM_ID = "default";
+ private static final String SOURCE_COMP = "source-spout";
+ private static final int SAMPLING_RATE = 1;
+ private static final double EWMA_FACTOR = 0.3;
+
+ @Mock private WorkerTopologyContext context;
+ @Mock private StormMetricRegistry metricRegistry;
+ @Mock private RateCounter rateCounter;
+ @Mock private RollingAverageGauge rollingAverageGauge;
+ @Mock private EwmaGauge ewmaGauge;
Review Comment:
These two `@Mock` fields are declared but never referenced anywhere in the test class — Mockito still spends init cost on them. Drop both.
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #8586:
URL: https://github.com/apache/storm/pull/8586#discussion_r3209376086
##########
storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.metrics2;
+
+import static org.apache.storm.utils.ConfigUtils.RFC1889_ALPHA;
+
+import com.codahale.metrics.Gauge;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Lock-free jitter estimator following RFC 1889 Section 6.3.1.
+ * The jitter accumulator is stored as raw IEEE 754 bits in an AtomicLong
+ * so that CAS can be used without locks.
+ * Thread safety: addValue is lock-free; getValue is wait-free.
+ */
+public class EwmaGauge implements Gauge<Double> {
+
+ private static final long UNSEEDED = Long.MIN_VALUE;
+ private static final long ZERO_BITS = Double.doubleToLongBits(0.0);
+
+ private final AtomicLong lastTransit = new AtomicLong(UNSEEDED);
+ private final AtomicLong jitterBits = new AtomicLong(ZERO_BITS);
+ private final double alpha;
+
+ EwmaGauge(double alpha) {
+ if (alpha <= 0.0 || alpha >= 1.0 || Double.isNaN(alpha)) {
+ throw new IllegalArgumentException(
+ "alpha must be in (0, 1), got: " + alpha);
+ }
+ this.alpha = alpha;
+ }
+
+ EwmaGauge() {
+ this(RFC1889_ALPHA); // 1.0 / 16.0
+ }
+
+ /**
+ * Update the jitter estimate.
+ *
+ * @param transitMs transit time for this tuple: {@code arrival - timestamp}
+ * Negative values are silently ignored.
+ */
+ public void addValue(long transitMs) {
+ if (transitMs < 0) {
+ return;
+ }
+
+ // Seed on the very first packet: store transit, nothing to diff against yet.
+ if (lastTransit.compareAndSet(UNSEEDED, transitMs)) {
+ return;
+ }
+
+ long prev = lastTransit.getAndSet(transitMs);
+ if (prev == UNSEEDED) {
+ // Lost a race during seeding; prev is not a real transit value.
+ return;
+ }
+
+ double d = Math.abs(transitMs - prev);
Review Comment:
Thank you for the explanation, I forgot about the CA2 pathology for Math.abs(Long.MIN_VALUE)
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "GGraziadei (via GitHub)" <gi...@apache.org>.
GGraziadei commented on code in PR #8586:
URL: https://github.com/apache/storm/pull/8586#discussion_r3209592323
##########
storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java:
##########
@@ -849,6 +849,23 @@ public void validateField(String name, Object o) {
}
}
+ public static class EwmaSmoothingFactorValidator extends Validator {
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ return;
+ }
+ if (o instanceof Number) {
+ double alpha = ((Number) o).doubleValue();
+ if (alpha > 0.0 && alpha < 1.0) {
+ return;
+ }
+ }
+ throw new IllegalArgumentException(
+ "Field " + name + " must be a number in the open interval (0, 1), got: " + o);
Review Comment:
Note: according to your comment and the symmetric implementation of `ObjectReader#getInt`
I added this branch to parse a double from a string in `ObjectReader#getDouble`.
```
...
else if (o instanceof String){
return Double.parseDouble((String) o);
}
...
```
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "rzo1 (via GitHub)" <gi...@apache.org>.
rzo1 commented on PR #8586:
URL: https://github.com/apache/storm/pull/8586#issuecomment-4385841547
**Question on the `d <= 0` short-circuit in `EwmaGauge.addValue`**
```java
if (d <= 0) {
return;
}
```
Is this skip intentional? Reading [RFC 3550 §A.8](https://datatracker.ietf.org/doc/html/rfc3550#appendix-A.8) (which supersedes RFC 1889 with the same text):
```c
d = transit - s->transit;
s->transit = transit;
if (d < 0) d = -d;
s->jitter += (1./16.) * ((double)d - s->jitter);
```
The update is unconditional. When `d == 0` it collapses to `J ← J · (1 − α)` ≈ `J · 15/16`, i.e. the EWMA decays toward zero. The RFC explicitly chooses 1/16 for its *"noise reduction ratio while maintaining a reasonable rate of convergence"* — convergence back toward zero during quiet periods is part of the spec's intent.
For comparison, the major RTP stacks all apply the update unconditionally:
- **GStreamer** — [`rtpsource.c:997`](https://github.com/GStreamer/gst-plugins-good/blob/master/gst/rtpmanager/rtpsource.c#L997)
```c
src->stats.jitter += diff - ((src->stats.jitter + 8) >> 4);
```
- **PJSIP** — [`rtcp.c:434`](https://github.com/pjsip/pjproject/blob/master/pjmedia/src/pjmedia/rtcp.c#L434)
```c
sess->jitter += d - ((sess->jitter + 8) >> 4);
```
- **WebRTC** — [`receive_statistics_impl.cc:165`](https://chromium.googlesource.com/external/webrtc/+/refs/heads/main/modules/rtp_rtcp/source/receive_statistics_impl.cc#165)
```cpp
int32_t jitter_diff_q4 = (time_diff_samples << 4) - jitter_q4_;
jitter_q4_ += ((jitter_diff_q4 + 8) >> 4);
```
(the only guard here is a `< 450000` anomaly cap, not a `d == 0` short-circuit)
The test `EwmaGaugeTest.zeroDeviationDecays` appears to lock in the current behavior, but its display name (*"Zero deviation decays jitter toward zero"*) suggests the original intent matched the spec while the assertion encodes the opposite. Worth double-checking which one was meant.
**Suggested fix:** drop the `if (d <= 0) return;` block — the CAS loop below is already correct for `d == 0`. (Optionally short-circuit the math with `updatedJitter = currentJitter * (1.0 - alpha)`, but you must still write it back.)
Was the skip intentional?
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Introduce jitter metric based on RFC 1889 Appendix A (storm)
Posted by "rzo1 (via GitHub)" <gi...@apache.org>.
rzo1 merged PR #8586:
URL: https://github.com/apache/storm/pull/8586
--
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: issues-unsubscribe@storm.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org