This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 7ee031e91 Introduce jitter metric based on RFC 1889 Appendix A (#8586)
7ee031e91 is described below
commit 7ee031e91e0760eb173794efaa35d904e18f5129
Author: Gianluca Graziadei <[email protected]>
AuthorDate: Mon May 18 21:18:46 2026 +0200
Introduce jitter metric based on RFC 1889 Appendix A (#8586)
* init
* allocate suppliers once
* improve jitter definition
* add documentation
* format
* format
* fix checkstyle
* fix RFC 1889 jitter definition, guarantee jitter decay if stable latency
* minor changes
---
conf/defaults.yaml | 48 +-
docs/Metrics.md | 42 +-
storm-client/src/jvm/org/apache/storm/Config.java | 102 +++--
.../jvm/org/apache/storm/metrics2/EwmaGauge.java | 84 ++++
.../jvm/org/apache/storm/metrics2/TaskMetrics.java | 89 +++-
.../jvm/org/apache/storm/utils/ConfigUtils.java | 23 +
.../jvm/org/apache/storm/utils/ObjectReader.java | 2 +
.../apache/storm/validation/ConfigValidation.java | 18 +
.../jvm/org/apache/storm/TestConfigValidate.java | 28 ++
.../org/apache/storm/metrics2/EwmaGaugeTest.java | 347 +++++++++++++++
.../org/apache/storm/metrics2/TaskMetricsTest.java | 485 +++++++++++++++++++++
11 files changed, 1195 insertions(+), 73 deletions(-)
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index 436809972..bc9a5979d 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -25,7 +25,7 @@ java.library.path:
"/usr/local/lib:/opt/local/lib:/usr/lib:/usr/lib64"
storm.local.dir: "storm-local"
storm.log4j2.conf.dir: "log4j2"
storm.zookeeper.servers:
- - "localhost"
+ - "localhost"
storm.zookeeper.port: 2181
storm.zookeeper.root: "/storm"
storm.zookeeper.session.timeout: 20000
@@ -52,7 +52,7 @@ storm.nimbus.retry.intervalceiling.millis: 60000
storm.nimbus.zookeeper.acls.check: true
storm.nimbus.zookeeper.acls.fixup: true
-storm.auth.simple-white-list.users: []
+storm.auth.simple-white-list.users: [ ]
storm.cluster.state.store: "org.apache.storm.cluster.ZKStateStorageFactory"
storm.meta.serialization.delegate:
"org.apache.storm.serialization.GzipThriftSerializationDelegate"
storm.codedistributor.class:
"org.apache.storm.codedistributor.LocalFileSystemCodeDistributor"
@@ -62,7 +62,7 @@ storm.health.check.timeout.ms: 5000
storm.disable.symlinks: false
### nimbus.* configs are for the master
-nimbus.seeds : ["localhost"]
+nimbus.seeds: [ "localhost" ]
nimbus.thrift.port: 6627
nimbus.thrift.threads: 64
nimbus.thrift.max_buffer_size: 1048576
@@ -163,10 +163,10 @@ storm.blobstore.acl.validation.enabled: false
### supervisor.* configs are for node supervisors
# Define the amount of workers that can be run on this machine. Each worker is
assigned a port to use for communication
supervisor.slots.ports:
- - 6700
- - 6701
- - 6702
- - 6703
+ - 6700
+ - 6701
+ - 6702
+ - 6703
supervisor.childopts: "-Xmx256m"
supervisor.run.worker.as.user: false
#how long supervisor will wait to ensure that a worker process is started
@@ -184,8 +184,8 @@ supervisor.worker.heartbeats.max.timeout.secs: 600
#For topology configurable heartbeat timeout, maximum allowed heartbeat
timeout.
worker.max.timeout.secs: 600
supervisor.enable: true
-supervisor.supervisors: []
-supervisor.supervisors.commands: []
+supervisor.supervisors: [ ]
+supervisor.supervisors.commands: [ ]
supervisor.memory.capacity.mb: 4096.0
#By convention 1 cpu core should be about 100, but this can be adjusted if
needed
# using 100 makes it simple to set the desired value to the capacity
measurement
@@ -278,6 +278,8 @@ 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
+topology.stats.ewma.smoothing.factor: 0.0625
topology.builtin.metrics.bucket.size.secs: 60
topology.fall.back.on.java.serialization: false
topology.worker.childopts: null
@@ -287,16 +289,16 @@ topology.worker.shared.thread.pool.size: 4
# Spout Wait Strategy - employed when there is no data to produce
topology.spout.wait.strategy: "org.apache.storm.policy.WaitStrategyProgressive"
-topology.spout.wait.park.microsec : 100 # park time for
org.apache.storm.policy.WaitStrategyPark. Busy spins if set to 0.
+topology.spout.wait.park.microsec: 100 # park time for
org.apache.storm.policy.WaitStrategyPark. Busy spins if set to 0.
topology.spout.wait.progressive.level1.count: 0 # number of
iterations to spend in level 1 [no sleep] of WaitStrategyProgressive, before
progressing to level 2
topology.spout.wait.progressive.level2.count: 0 # number of
iterations to spend in level 2 [parkNanos(1)] of WaitStrategyProgressive,
before progressing to level 3
topology.spout.wait.progressive.level3.sleep.millis: 1 # sleep duration for
idling iterations in level 3 of WaitStrategyProgressive
# Bolt Wait Strategy - employed when there is no data in its receive buffer to
process
-topology.bolt.wait.strategy : "org.apache.storm.policy.WaitStrategyProgressive"
+topology.bolt.wait.strategy: "org.apache.storm.policy.WaitStrategyProgressive"
-topology.bolt.wait.park.microsec : 100 # park time for
org.apache.storm.policy.WaitStrategyPark. Busy spins if set to 0.
+topology.bolt.wait.park.microsec: 100 # park time for
org.apache.storm.policy.WaitStrategyPark. Busy spins if set to 0.
topology.bolt.wait.progressive.level1.count: 1 # number of iterations
to spend in level 1 [no sleep] of WaitStrategyProgressive, before progressing
to level 2
topology.bolt.wait.progressive.level2.count: 1000 # number of iterations
to spend in level 2 [parkNanos(1)] of WaitStrategyProgressive, before
progressing to level 3
@@ -363,7 +365,7 @@
blacklist.scheduler.assume.supervisor.bad.based.on.bad.slot: true
dev.zookeeper.path: "/tmp/dev-storm-zookeeper"
-pacemaker.servers: []
+pacemaker.servers: [ ]
pacemaker.port: 6699
pacemaker.base.threads: 10
pacemaker.max.threads: 50
@@ -371,12 +373,12 @@ pacemaker.client.max.threads: 2
pacemaker.thread.timeout: 10
pacemaker.childopts: "-Xmx1024m"
pacemaker.auth.method: "NONE"
-pacemaker.kerberos.users: []
+pacemaker.kerberos.users: [ ]
pacemaker.thrift.message.size.max: 10485760
#default storm daemon metrics reporter plugins
storm.daemon.metrics.reporter.plugins:
- - "org.apache.storm.daemon.metrics.reporters.JmxPreparableReporter"
+ - "org.apache.storm.daemon.metrics.reporters.JmxPreparableReporter"
storm.daemon.metrics.reporter.interval.secs: 10
storm.metricstore.class: "org.apache.storm.metricstore.rocksdb.RocksDbStore"
@@ -399,8 +401,8 @@ storm.cgroup.inherit.cpuset.configs: false
# Configs for CGroup support
storm.cgroup.hierarchy.dir: "/cgroup/storm_resources"
storm.cgroup.resources:
- - "cpu"
- - "memory"
+ - "cpu"
+ - "memory"
storm.cgroup.hierarchy.name: "storm"
storm.supervisor.cgroup.rootdir: "storm"
storm.cgroup.cgexec.cmd: "/bin/cgexec"
@@ -419,12 +421,12 @@ storm.worker.min.cpu.pcore.percent: 0.0
storm.topology.classpath.beginning.enabled: false
worker.metrics:
- "CGroupMemory": "org.apache.storm.metrics2.cgroup.CGroupMemoryUsage"
- "CGroupMemoryLimit": "org.apache.storm.metrics2.cgroup.CGroupMemoryLimit"
- "CGroupCpu": "org.apache.storm.metrics2.cgroup.CGroupCpu"
- "CGroupCpuGuarantee": "org.apache.storm.metrics2.cgroup.CGroupCpuGuarantee"
- "CGroupCpuGuaranteeByCfsQuota":
"org.apache.storm.metrics2.cgroup.CGroupCpuGuaranteeByCfsQuota"
- "CGroupCpuStat": "org.apache.storm.metrics2.cgroup.CGroupCpuStat"
+ "CGroupMemory": "org.apache.storm.metrics2.cgroup.CGroupMemoryUsage"
+ "CGroupMemoryLimit": "org.apache.storm.metrics2.cgroup.CGroupMemoryLimit"
+ "CGroupCpu": "org.apache.storm.metrics2.cgroup.CGroupCpu"
+ "CGroupCpuGuarantee": "org.apache.storm.metrics2.cgroup.CGroupCpuGuarantee"
+ "CGroupCpuGuaranteeByCfsQuota":
"org.apache.storm.metrics2.cgroup.CGroupCpuGuaranteeByCfsQuota"
+ "CGroupCpuStat": "org.apache.storm.metrics2.cgroup.CGroupCpuStat"
# The number of buckets for running statistics
num.stat.buckets: 20
diff --git a/docs/Metrics.md b/docs/Metrics.md
index 580d66e9b..8d620f0da 100644
--- a/docs/Metrics.md
+++ b/docs/Metrics.md
@@ -180,15 +180,22 @@ Similar to the tuple counting metrics storm also collects
average latency metric
##### `__complete-latency`
-The complete latency is just for spouts. It is the average amount of time it
took for `ack` or `fail` to be called for a tuple after it was emitted. If
acking is disabled this metric is likely to be blank or 0 for all values, and
should be ignored.
+The complete latency is just for spouts. It is the average amount of time it
took for `ack` or `fail` to be called for a
+tuple after it was emitted. If acking is disabled this metric is likely to be
blank or 0 for all values, and should be
+ignored.
##### `__execute-latency`
-This is just for bolts. It is the average amount of time that the bolt spent
in the call to the `execute` method. The higher this gets, the lower the
throughput of tuples per bolt instance.
+This is just for bolts. It is the average amount of time that the bolt spent
in the call to the `execute` method. The
+higher this gets, the lower the throughput of tuples per bolt instance.
##### `__process-latency`
-This is also just for bolts. It is the average amount of time between when
`execute` was called to start processing a tuple, to when it was acked or
failed by the bolt. If your bolt is a very simple bolt and the processing is
synchronous then `__process-latency` and `__execute-latency` should be very
close to one another, with process latency being slightly smaller. If you are
doing a join or have asynchronous processing then it may take a while for a
tuple to be acked so the process [...]
+This is also just for bolts. It is the average amount of time between when
`execute` was called to start processing a
+tuple, to when it was acked or failed by the bolt. If your bolt is a very
simple bolt and the processing is synchronous
+then `__process-latency` and `__execute-latency` should be very close to one
another, with process latency being
+slightly smaller. If you are doing a join or have asynchronous processing then
it may take a while for a tuple to be
+acked so the process latency would be higher than the execute latency.
##### `__skipped-max-spout-ms`
@@ -207,6 +214,35 @@ This metric indicates the overflow count last time BP
status was sent, with a mi
This metric records how much time a spout was idle because the topology was
deactivated. This is the total time in milliseconds, not the average amount of
time and is not sub-sampled.
+#### Tuple Jitter Metrics
+To activate jitter-based metrics, `topology.stats.ewma.enable` must be set to
`true`, which switches the system to a jitter estimation based on an
Exponential Moving Average (EWMA).
+In this model, jitter is dynamically updated by weighting new latency samples
against historical data using a smoothing factor
(`topology.stats.ewma.smoothing.factor`).
+This parameter, which defaults to 0.0625 (equivalent to $1/16$ or a 4-bit
right shift), determines the metric's reactivity: higher values make the jitter
more sensitive to recent spikes, while lower values prioritize long-term
stability.
+Operators should be aware that enabling this feature triples the gauge count
for every component-stream pair per task; this significant increase in metric
cardinality can impact TSDB storage and costs, so backend capacity should be
verified before deployment.
+
+##### `__complete-jitter`
+
+This metric is specific to spouts. It measures the variation (jitter) in the
total completion time (end-to-end latency)
+of tuples, calculated using the exponentially weighted moving average (EWMA)
algorithm as defined in RFC 1889 §A.8 / RFC 3550 §A.8.
+While `__complete-latency` indicates the average amount of time it took for a
tuple to be fully processed by the
+topology (from emission to the final ack), the jitter metric quantifies the
consistency of that process. If acking is
+disabled, this metric is likely to be blank or 0 and should be ignored.
+
+##### `__execute-jitter`
+
+This metric is specific to bolts. It measures the variation (jitter) in the
time spent within the execute method,
+calculated using the exponentially weighted moving average (EWMA) algorithm as
defined in RFC 1889 §A.8 / RFC 3550 §A.8.
+While `__execute-latency` provides the average time spent in the execute call,
the jitter metric quantifies the
+predictability of that execution time. It is a critical indicator of
computational "smoothness".
+
+##### `__process-jitter`
+
+This metric is specific to bolts. It measures the variation (jitter) in the
process latency, calculated using the
+exponentially weighted moving average (EWMA) algorithm as defined in RFC 1889
§A.8 / RFC 3550 §A.8.
+While `__process-latency` provides the average time a tuple spends being
processed, the jitter metric quantifies the
+stability of that processing time. It helps identify "noisy" execution
environments where processing times fluctuate
+significantly, even if the average remains within acceptable limits.
+
#### Error Reporting Metrics
Storm also collects error reporting metrics for bolts and spouts.
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index e332b726b..62770b62e 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -350,13 +350,13 @@ public class Config extends HashMap<String, Object> {
* <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most
"2" comp-1 on same node</p>
* <p>comp-2 and comp-4 cannot be on same worker (missing comp-1 is
implied from comp-1 constraint)</p>
*
- * <p>
- * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents":
["comp-2", "comp-3" ] },
- * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
- * }
- * </p>
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents":
["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
*/
- @IsExactlyOneOf(valueValidatorClasses = {
ListOfListOfStringValidator.class, RasConstraintsTypeValidator.class })
+ @IsExactlyOneOf(valueValidatorClasses =
{ListOfListOfStringValidator.class, RasConstraintsTypeValidator.class})
public static final String TOPOLOGY_RAS_CONSTRAINTS =
"topology.ras.constraints";
/**
@@ -424,17 +424,17 @@ public class Config extends HashMap<String, Object> {
*
* <p>
* 1. If not setting this variable or setting it as null,
- * a. If RAS is not used:
- * Nimbus will set it to {@link Config#TOPOLOGY_WORKERS}.
- * b. If RAS is used:
- * Nimbus will set it to (the estimate number of workers * {@link
Config#TOPOLOGY_RAS_ACKER_EXECUTORS_PER_WORKER}).
- * {@link Config#TOPOLOGY_RAS_ACKER_EXECUTORS_PER_WORKER} is
default to be 1 if not set.
+ * a. If RAS is not used:
+ * Nimbus will set it to {@link Config#TOPOLOGY_WORKERS}.
+ * b. If RAS is used:
+ * Nimbus will set it to (the estimate number of workers * {@link
Config#TOPOLOGY_RAS_ACKER_EXECUTORS_PER_WORKER}).
+ * {@link Config#TOPOLOGY_RAS_ACKER_EXECUTORS_PER_WORKER} is default to be
1 if not set.
* 2. If this variable is set to 0,
- * then Storm will immediately ack tuples as soon as they come off the
spout,
- * effectively disabling reliability.
+ * then Storm will immediately ack tuples as soon as they come off the
spout,
+ * effectively disabling reliability.
* 3. If this variable is set to a positive integer,
- * Storm will not honor {@link
Config#TOPOLOGY_RAS_ACKER_EXECUTORS_PER_WORKER} setting.
- * Instead, nimbus will set it as (this variable / estimate num of
workers).
+ * Storm will not honor {@link
Config#TOPOLOGY_RAS_ACKER_EXECUTORS_PER_WORKER} setting.
+ * Instead, nimbus will set it as (this variable / estimate num of
workers).
* </p>
*/
@IsInteger
@@ -465,7 +465,7 @@ public class Config extends HashMap<String, Object> {
* <p>Note that EventLoggerBolt takes care of all the implementations of
IEventLogger, hence registering many
* implementations (especially they're implemented as 'blocking' manner)
would slow down overall topology.
*/
- @IsListEntryCustom(entryValidatorClasses = {
EventLoggerRegistryValidator.class })
+ @IsListEntryCustom(entryValidatorClasses =
{EventLoggerRegistryValidator.class})
public static final String TOPOLOGY_EVENT_LOGGER_REGISTER =
"topology.event.logger.register";
/**
* How many executors to spawn for event logger.
@@ -543,7 +543,7 @@ public class Config extends HashMap<String, Object> {
* it's parallelism is configurable.
*/
- @IsListEntryCustom(entryValidatorClasses = { MetricRegistryValidator.class
})
+ @IsListEntryCustom(entryValidatorClasses = {MetricRegistryValidator.class})
public static final String TOPOLOGY_METRICS_CONSUMER_REGISTER =
"topology.metrics.consumer.register";
/**
* Enable tracking of network message byte counts per source-destination
task. This is off by default as it creates tasks^2 metric
@@ -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 1889 §A.8).
+ *
+ * @see <a href="https://www.rfc-editor.org/rfc/rfc1889#appendix-A.8";>RFC
1889 §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 1889 §A.8). The default value is set to 1/16.
+ *
+ * @see <a href="https://www.rfc-editor.org/rfc/rfc1889#appendix-A.8";>RFC
1889 §A.8</a>
+ */
+ @CustomValidator(validatorClass =
ConfigValidation.EwmaSmoothingFactorValidator.class)
+ public static final String TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR =
"topology.stats.ewma.smoothing.factor";
/**
* The time period that builtin metrics data in bucketed into.
*/
@@ -833,14 +847,14 @@ public class Config extends HashMap<String, Object> {
* Topology central logging sensitivity to determine who has access to
logs in central logging system. The possible values are: S0 -
* Public (open to all users on grid) S1 - Restricted S2 - Confidential S3
- Secret (default.)
*/
- @IsString(acceptedValues = { "S0", "S1", "S2", "S3" })
+ @IsString(acceptedValues = {"S0", "S1", "S2", "S3"})
public static final String TOPOLOGY_LOGGING_SENSITIVITY =
"topology.logging.sensitivity";
/**
* Log file the user can use to configure Log4j2.
* Can be a resource in the jar (specified with
classpath:/path/to/resource) or a file.
* This configuration is applied in addition to the regular worker log4j2
configuration.
* The configs are merged according to the rules here:
- *
https://logging.apache.org/log4j/2.x/manual/configuration.html#CompositeConfiguration
+ *
https://logging.apache.org/log4j/2.x/manual/configuration.html#CompositeConfiguration
*/
@IsString
public static final String TOPOLOGY_LOGGING_CONFIG_FILE =
"topology.logging.config";
@@ -884,7 +898,8 @@ public class Config extends HashMap<String, Object> {
* Alternatively set {@code storm.scheduler} to {@code
org.apache.storm.scheduler.resource.ResourceAwareScheduler}
* using {@link Config#TOPOLOGY_SCHEDULER_STRATEGY} set to
* {@code
org.apache.storm.scheduler.resource.strategies.scheduling.RoundRobinResourceAwareStrategy}
- * */
+ *
+ */
@IsInteger
@IsPositiveNumber
public static final String TOPOLOGY_ISOLATED_MACHINES =
"topology.isolate.machines";
@@ -1434,22 +1449,34 @@ public class Config extends HashMap<String, Object> {
@IsString
public static final String STORM_ZOOKEEPER_TOPOLOGY_AUTH_SCHEME =
"storm.zookeeper.topology.auth.scheme";
- /** Enable SSL/TLS for ZooKeeper client connection. */
+ /**
+ * Enable SSL/TLS for ZooKeeper client connection.
+ */
@IsBoolean
public static final String ZK_SSL_ENABLE = "storm.zookeeper.ssl.enable";
- /** Keystore location for ZooKeeper client connection over SSL. */
+ /**
+ * Keystore location for ZooKeeper client connection over SSL.
+ */
@IsString
public static final String STORM_ZOOKEEPER_SSL_KEYSTORE_PATH =
"storm.zookeeper.ssl.keystore.path";
- /** Keystore password for ZooKeeper client connection over SSL. */
+ /**
+ * Keystore password for ZooKeeper client connection over SSL.
+ */
@IsString
public static final String STORM_ZOOKEEPER_SSL_KEYSTORE_PASSWORD =
"storm.zookeeper.ssl.keystore.password";
- /** Truststore location for ZooKeeper client connection over SSL. */
+ /**
+ * Truststore location for ZooKeeper client connection over SSL.
+ */
@IsString
public static final String STORM_ZOOKEEPER_SSL_TRUSTSTORE_PATH =
"storm.zookeeper.ssl.truststore.path";
- /** Truststore password for ZooKeeper client connection over SSL. */
+ /**
+ * Truststore password for ZooKeeper client connection over SSL.
+ */
@IsString
public static final String STORM_ZOOKEEPER_SSL_TRUSTSTORE_PASSWORD =
"storm.zookeeper.ssl.truststore.password";
- /** Enable or disable hostname verification.*/
+ /**
+ * Enable or disable hostname verification.
+ */
@IsBoolean
public static final String STORM_ZOOKEEPER_SSL_HOSTNAME_VERIFICATION =
"storm.zookeeper.ssl.hostnameVerification";
/**
@@ -1462,13 +1489,13 @@ public class Config extends HashMap<String, Object> {
/**
* Configure the topology metrics reporters to be used on workers.
*/
- @IsListEntryCustom(entryValidatorClasses = {
MetricReportersValidator.class })
+ @IsListEntryCustom(entryValidatorClasses =
{MetricReportersValidator.class})
public static final String TOPOLOGY_METRICS_REPORTERS =
"topology.metrics.reporters";
/**
* A list of system metrics reporters that will get added to each topology.
*/
- @IsListEntryCustom(entryValidatorClasses = {
MetricReportersValidator.class })
+ @IsListEntryCustom(entryValidatorClasses =
{MetricReportersValidator.class})
public static final String STORM_TOPOLOGY_METRICS_SYSTEM_REPORTERS =
"storm.topology.metrics.system.reporters";
/**
@@ -1476,7 +1503,7 @@ public class Config extends HashMap<String, Object> {
* Use {@link Config#TOPOLOGY_METRICS_REPORTERS} instead.
*/
@Deprecated(forRemoval = true, since = "2.0.0")
- @IsListEntryCustom(entryValidatorClasses = {
MetricReportersValidator.class })
+ @IsListEntryCustom(entryValidatorClasses =
{MetricReportersValidator.class})
public static final String STORM_METRICS_REPORTERS =
"storm.metrics.reporters";
/**
@@ -1511,6 +1538,7 @@ public class Config extends HashMap<String, Object> {
public static final String BLOBSTORE_HDFS_PRINCIPAL =
"blobstore.hdfs.principal";
/**
* keytab for nimbus/supervisor to use to access secure hdfs for the
blobstore.
+ *
* @Deprecated Use {@link Config#STORM_HDFS_LOGIN_KEYTAB} instead.
*/
@Deprecated
@@ -1753,7 +1781,7 @@ public class Config extends HashMap<String, Object> {
*/
@IsInteger
public static final String STORM_MESSAGING_NETTY_CLIENT_WORKER_THREADS =
- "storm.messaging.netty.client_worker_threads";
+ "storm.messaging.netty.client_worker_threads";
/**
* Netty based messaging: Enables TLS connections between workers.
@@ -1808,7 +1836,7 @@ public class Config extends HashMap<String, Object> {
*/
@IsString
public static final String
STORM_MESSAGING_NETTY_TLS_CLIENT_TRUSTSTORE_PASSWORD =
- "storm.messaging.netty.tls.client.truststore.password";
+ "storm.messaging.netty.tls.client.truststore.password";
/**
* Netty based messaging: Specifies the client keystore when TLS is
enabled.
@@ -1821,7 +1849,7 @@ public class Config extends HashMap<String, Object> {
*/
@IsString
public static final String
STORM_MESSAGING_NETTY_TLS_CLIENT_KEYSTORE_PASSWORD =
- "storm.messaging.netty.tls.client.keystore.password";
+ "storm.messaging.netty.tls.client.keystore.password";
/**
* Netty based messaging: Specifies the protocols TLS is enabled.
@@ -1830,7 +1858,7 @@ public class Config extends HashMap<String, Object> {
public static final String STORM_MESSAGING_NETTY_TLS_SSL_PROTOCOLS =
"storm.messaging.netty.tls.ssl.protocols";
/**
- /**
+ * /**
* Netty based messaging: The number of milliseconds that a Netty client
will retry flushing messages that are already
* buffered to be sent.
*/
@@ -1915,7 +1943,7 @@ public class Config extends HashMap<String, Object> {
@IsPositiveNumber
@IsInteger
public static final String
STORM_BLOBSTORE_DEPENDENCY_JAR_UPLOAD_CHUNK_SIZE_BYTES =
- "storm.blobstore.dependency.jar.upload.chunk.size.bytes";
+ "storm.blobstore.dependency.jar.upload.chunk.size.bytes";
/**
* FQCN of a class that implements {@code ISubmitterHook} @see
ISubmitterHook for details.
*/
@@ -1924,8 +1952,8 @@ public class Config extends HashMap<String, Object> {
/**
* Impersonation user ACL config entries.
*/
- @IsMapEntryCustom(keyValidatorClasses = {
ConfigValidation.StringValidator.class },
- valueValidatorClasses = {
ConfigValidation.ImpersonationAclUserEntryValidator.class })
+ @IsMapEntryCustom(keyValidatorClasses =
{ConfigValidation.StringValidator.class},
+ valueValidatorClasses =
{ConfigValidation.ImpersonationAclUserEntryValidator.class})
public static final String NIMBUS_IMPERSONATION_ACL =
"nimbus.impersonation.acl";
/**
* A whitelist of the RAS scheduler strategies allowed by nimbus. Should
be a list of fully-qualified class names or null to allow all.
@@ -2430,6 +2458,7 @@ public class Config extends HashMap<String, Object> {
/**
* Get the hostname substituted hdfs principal.
+ *
* @param conf the storm Configuration
* @return the principal
* @throws UnknownHostException on UnknowHostException
@@ -2458,6 +2487,7 @@ public class Config extends HashMap<String, Object> {
/**
* Get the hdfs keytab.
+ *
* @param conf the storm Configuration
* @return the keytab
*/
diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java
b/storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java
new file mode 100644
index 000000000..857e34215
--- /dev/null
+++ b/storm-client/src/jvm/org/apache/storm/metrics2/EwmaGauge.java
@@ -0,0 +1,84 @@
+/**
+ * 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.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 §A.8 / RFC 3550 §A.8.
+ * 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);
+ // Safe from Math.abs(Long.MIN_VALUE) pathology: both transitMs and
prev
+ // are >= 0 (enforced by the negative-guard), so their
+ // difference is in [-Long.MAX_VALUE, Long.MAX_VALUE].
+ 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());
+ }
+}
diff --git a/storm-client/src/jvm/org/apache/storm/metrics2/TaskMetrics.java
b/storm-client/src/jvm/org/apache/storm/metrics2/TaskMetrics.java
index 0ac3a5e94..78cd6d3b9 100644
--- a/storm-client/src/jvm/org/apache/storm/metrics2/TaskMetrics.java
+++ b/storm-client/src/jvm/org/apache/storm/metrics2/TaskMetrics.java
@@ -12,10 +12,12 @@
package org.apache.storm.metrics2;
-import com.codahale.metrics.Counter;
+import com.codahale.metrics.Gauge;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.function.Supplier;
+
import org.apache.storm.task.WorkerTopologyContext;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.Utils;
@@ -27,12 +29,18 @@ 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_JITTER =
"__process-jitter";
private static final String METRIC_NAME_COMPLETE_LATENCY =
"__complete-latency";
+ private static final String METRIC_NAME_COMPLETE_JITTER =
"__complete-jitter";
private static final String METRIC_NAME_EXECUTE_LATENCY =
"__execute-latency";
+ private static final String METRIC_NAME_EXECUTE_JITTER =
"__execute-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<>();
+ // Gauge supplier singleton factories
+ private final Supplier<EwmaGauge> ewmaGaugeFactory;
+ private final Supplier<RollingAverageGauge> rollingAverageGaugeFactory;
private final String topologyId;
private final String componentId;
@@ -40,6 +48,7 @@ public class TaskMetrics {
private final Integer workerPort;
private final StormMetricRegistry metricRegistry;
private final int samplingRate;
+ private final boolean ewmaEnable;
public TaskMetrics(WorkerTopologyContext context, String componentId,
Integer taskid,
@@ -50,6 +59,10 @@ public class TaskMetrics {
this.taskId = taskid;
this.workerPort = context.getThisWorkerPort();
this.samplingRate = ConfigUtils.samplingRate(topoConf);
+ double ewmaSmoothingFactor = ConfigUtils.ewmaSmoothingFactor(topoConf);
+ this.ewmaEnable = ConfigUtils.ewmaEnable(topoConf);
+ this.rollingAverageGaugeFactory = RollingAverageGauge::new;
+ this.ewmaGaugeFactory = () -> new EwmaGauge(ewmaSmoothingFactor);
}
public void setCapacity(double capacity) {
@@ -67,6 +80,12 @@ public class TaskMetrics {
metricName = METRIC_NAME_COMPLETE_LATENCY + "-" + streamId;
RollingAverageGauge gauge = this.getRollingAverageGauge(metricName,
streamId);
gauge.addValue(latencyMs);
+
+ if (this.ewmaEnable) {
+ metricName = METRIC_NAME_COMPLETE_JITTER + "-" + streamId;
+ EwmaGauge ewmaGauge =
this.getExponentialWeightedMovingAverageGauge(metricName, streamId);
+ ewmaGauge.addValue(latencyMs);
+ }
}
public void boltAckedTuple(String sourceComponentId, String
sourceStreamId, long latencyMs) {
@@ -78,6 +97,12 @@ public class TaskMetrics {
metricName = METRIC_NAME_PROCESS_LATENCY + "-" + key;
RollingAverageGauge gauge = this.getRollingAverageGauge(metricName,
sourceStreamId);
gauge.addValue(latencyMs);
+
+ if (this.ewmaEnable) {
+ metricName = METRIC_NAME_PROCESS_JITTER + "-" + key;
+ EwmaGauge ewmaGauge =
this.getExponentialWeightedMovingAverageGauge(metricName, sourceStreamId);
+ ewmaGauge.addValue(latencyMs);
+ }
}
public void spoutFailedTuple(String streamId) {
@@ -117,6 +142,12 @@ public class TaskMetrics {
metricName = METRIC_NAME_EXECUTE_LATENCY + "-" + key;
RollingAverageGauge gauge = this.getRollingAverageGauge(metricName,
sourceStreamId);
gauge.addValue(latencyMs);
+
+ if (this.ewmaEnable) {
+ metricName = METRIC_NAME_EXECUTE_JITTER + "-" + key;
+ EwmaGauge ewmaGauge =
this.getExponentialWeightedMovingAverageGauge(metricName, sourceStreamId);
+ ewmaGauge.addValue(latencyMs);
+ }
}
private RateCounter getRateCounter(String metricName, String streamId) {
@@ -135,18 +166,54 @@ public class TaskMetrics {
}
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 getExponentialWeightedMovingAverageGauge(String
metricName, String streamId) {
+ return getOrCreateGauge(metricName, streamId, EwmaGauge.class,
this.ewmaGaugeFactory);
+ }
+
+ private <G extends Gauge<?>> G getOrCreateGauge(
+ String metricName,
+ String streamId,
+ Class<G> gaugeClass,
+ Supplier<G> factory) {
+
+ Object existing = this.gauges.get(metricName);
+ if (existing == null) {
synchronized (this) {
- gauge = this.gauges.get(metricName);
- if (gauge == null) {
- gauge = new RollingAverageGauge();
- metricRegistry.gauge(metricName, gauge, this.topologyId,
this.componentId,
- streamId, this.taskId, this.workerPort);
- this.gauges.put(metricName, gauge);
+ existing = this.gauges.get(metricName);
+ if (existing == null) {
+ G created = factory.get();
+ registerGauge(metricName, streamId, created);
+ this.gauges.put(metricName, created);
+ return created;
}
}
}
- return gauge;
+
+ if (!gaugeClass.isInstance(existing)) {
+ throw new IllegalStateException(
+ "Metric '" + metricName + "' is registered as "
+ + existing.getClass().getName()
+ + " but expected " + gaugeClass.getName());
+ }
+
+ return gaugeClass.cast(existing);
+ }
+
+ /*
+ * Safe cast: G is bounded by Gauge<?> in the signature of
getOrCreateGauge,
+ * so every instance of G is by definition a Gauge.
+ * The cast to raw Gauge is required because metricRegistry.gauge() does
not
+ * accept Gauge<?> the wildcard is not compatible with the type parameter T
+ * expected by the external API. Type-safety is guaranteed by the bound
+ * <G extends Gauge<?>> declared at the call site.
+ */
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private void registerGauge(String metricName, String streamId, Gauge<?>
gauge) {
+ metricRegistry.gauge(metricName, (Gauge) gauge, this.topologyId,
+ this.componentId, streamId, this.taskId, this.workerPort);
}
+
}
diff --git a/storm-client/src/jvm/org/apache/storm/utils/ConfigUtils.java
b/storm-client/src/jvm/org/apache/storm/utils/ConfigUtils.java
index 9357cc562..94599ec89 100644
--- a/storm-client/src/jvm/org/apache/storm/utils/ConfigUtils.java
+++ b/storm-client/src/jvm/org/apache/storm/utils/ConfigUtils.java
@@ -39,6 +39,7 @@ public class ConfigUtils {
public static final String FILE_SEPARATOR = File.separator;
public static final String STORM_HOME = "storm.home";
public static final String RESOURCES_SUBDIR = "resources";
+ public static final double RFC1889_ALPHA = 1.0 / 16.0;
private static final Set<String> passwordConfigKeys = new HashSet<>();
@@ -175,6 +176,28 @@ public class ConfigUtils {
throw new IllegalArgumentException("Illegal topology.stats.sample.rate
in conf: " + rate);
}
+ public static double ewmaSmoothingFactor(Map<String, Object> conf) {
+ Object value = conf.get(Config.TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR);
+ if (value == null) {
+ return RFC1889_ALPHA;
+ }
+ double alpha = ObjectReader.getDouble(value);
+ if (alpha > 0.0 && alpha < 1.0) {
+ return alpha;
+ }
+ throw new IllegalArgumentException(
+ "Illegal " + Config.TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR
+ + " in conf: " + alpha + " must be in (0, 1)");
+ }
+
+ public static boolean ewmaEnable(Map<String, Object> conf) {
+ Object value = conf.get(Config.TOPOLOGY_STATS_EWMA_ENABLE);
+ if (value == null) {
+ return false;
+ }
+ return ObjectReader.getBoolean(value, false);
+ }
+
public static BooleanSupplier mkStatsSampler(Map<String, Object> conf) {
return evenSampler(samplingRate(conf));
}
diff --git a/storm-client/src/jvm/org/apache/storm/utils/ObjectReader.java
b/storm-client/src/jvm/org/apache/storm/utils/ObjectReader.java
index 54445b5eb..ac28be8cf 100644
--- a/storm-client/src/jvm/org/apache/storm/utils/ObjectReader.java
+++ b/storm-client/src/jvm/org/apache/storm/utils/ObjectReader.java
@@ -128,6 +128,8 @@ public class ObjectReader {
}
if (o instanceof Number) {
return ((Number) o).doubleValue();
+ } else if (o instanceof String) {
+ return Double.parseDouble((String) o);
} else {
throw new IllegalArgumentException("Don't know how to convert (" +
o + ") to double");
}
diff --git
a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
index 4175ee9f4..0206ec458 100644
--- a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
+++ b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
@@ -30,6 +30,7 @@ import java.util.Set;
import java.util.stream.Collectors;
import org.apache.storm.Config;
+import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Utils;
import org.apache.storm.validation.ConfigValidationAnnotations.ValidatorParams;
import org.slf4j.Logger;
@@ -849,6 +850,23 @@ public class ConfigValidation {
}
}
+ public static class EwmaSmoothingFactorValidator extends Validator {
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ return;
+ }
+ // ObjectReader.getDouble(o) handles the type conversion and will
throw an
+ // IllegalArgumentException if the value cannot be parsed as a
number.
+ double alpha = ObjectReader.getDouble(o);
+ 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);
+ }
+ }
+
public static class CustomIsExactlyOneOfValidators extends Validator {
private Class<?>[] subValidators;
private List<String> validatorClassNames;
diff --git a/storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
b/storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
index 3c335308d..31b9c4834 100644
--- a/storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
+++ b/storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
@@ -206,6 +206,34 @@ public class TestConfigValidate {
ConfigValidation.validateFields(conf);
}
+ @Test
+ public void testTopologyStatsEwmaEnableIsBoolean() {
+ Map<String, Object> conf = new HashMap<>();
+ // optional configuration
+ ConfigValidation.validateFields(conf);
+ conf.put(Config.TOPOLOGY_STATS_EWMA_ENABLE, true);
+ ConfigValidation.validateFields(conf);
+ conf.put(Config.TOPOLOGY_STATS_EWMA_ENABLE, false);
+ ConfigValidation.validateFields(conf);
+ }
+
+ @Test
+ public void testTopologyStatsEwmaSmoothingFactorCustomValidator() {
+ Map<String, Object> conf = new HashMap<>();
+ // optional configuration
+ ConfigValidation.validateFields(conf);
+ conf.put(Config.TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR, 0.1);
+ ConfigValidation.validateFields(conf);
+ conf.put(Config.TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR, "0.1");
+ ConfigValidation.validateFields(conf);
+ conf.put(Config.TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR, 0.9);
+ ConfigValidation.validateFields(conf);
+ for (Object notAllowedValue : new Object[]{0.0, -0.1, 1.9, "1.9"}) {
+ conf.put(Config.TOPOLOGY_STATS_EWMA_SMOOTHING_FACTOR,
notAllowedValue);
+ assertThrows(IllegalArgumentException.class, () ->
ConfigValidation.validateFields(conf));
+ }
+ }
+
@Test
public void testWorkerChildoptsIsStringOrStringList() {
Map<String, Object> conf = new HashMap<>();
diff --git a/storm-client/test/jvm/org/apache/storm/metrics2/EwmaGaugeTest.java
b/storm-client/test/jvm/org/apache/storm/metrics2/EwmaGaugeTest.java
new file mode 100644
index 000000000..ed1b5009c
--- /dev/null
+++ b/storm-client/test/jvm/org/apache/storm/metrics2/EwmaGaugeTest.java
@@ -0,0 +1,347 @@
+/**
+ * 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.metrics2;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class EwmaGaugeTest {
+
+ private static final double DELTA = 1e-9;
+
+ @Nested
+ @DisplayName("Construction")
+ class ConstructionTest {
+
+ @Test
+ @DisplayName("Default constructor uses RFC 1889 alpha (1/16)")
+ void defaultAlpha() {
+ EwmaGauge gauge = new EwmaGauge();
+ gauge.addValue(0L);
+ gauge.addValue(16L); // D = 16 ; J = 0 + (16 - 0) * (1/16) = 1.0
+ assertEquals(1.0, gauge.getValue(), DELTA);
+ }
+
+ @Test
+ @DisplayName("Invalid alpha values throw IllegalArgumentException")
+ void invalidAlphaThrows() {
+ double[] invalidAlphas = {
+ 0.0, 1.0, -0.1, 1.1,
+ Double.NaN, Double.POSITIVE_INFINITY,
Double.NEGATIVE_INFINITY
+ };
+ for (double alpha : invalidAlphas) {
+ assertThrows(IllegalArgumentException.class,
+ () -> new EwmaGauge(alpha),
+ "Expected IllegalArgumentException for alpha=" +
alpha);
+ }
+ }
+
+ @Test
+ @DisplayName("Valid alpha boundary values are accepted")
+ void validAlphaAccepted() {
+ double[] validAlphas = {0.001, 0.0625, 0.5, 0.999};
+ for (double alpha : validAlphas) {
+ assertNotNull(new EwmaGauge(alpha),
+ "Expected no exception for alpha=" + alpha);
+ }
+ }
+ }
+
+
+ @Nested
+ @DisplayName("Cold-start semantics")
+ class ColdStartTest {
+
+ private EwmaGauge gauge;
+
+ @BeforeEach
+ void setUp() {
+ gauge = new EwmaGauge();
+ }
+
+ @Test
+ @DisplayName("getValue() returns 0.0 before any sample")
+ void noSamples() {
+ assertEquals(0.0, gauge.getValue(), DELTA);
+ }
+
+ @Test
+ @DisplayName("getValue() returns 0.0 after exactly one sample (seed
only)")
+ void oneSample() {
+ gauge.addValue(100L);
+ assertEquals(0.0, gauge.getValue(), DELTA);
+ }
+
+ }
+
+ @Nested
+ @DisplayName("EWMA formula RFC 1889 §A.8")
+ class FormulaTest {
+
+ @Test
+ @DisplayName("Single update: J = 0 + (D - 0) * alpha")
+ void singleDeviation() {
+ EwmaGauge gauge = new EwmaGauge(0.5);
+ gauge.addValue(0L);
+ gauge.addValue(10L);
+ assertEquals(5.0, gauge.getValue(), DELTA);
+ }
+
+ @Test
+ @DisplayName("Manual step-by-step verification against reference
values")
+ void manualSteps() {
+ EwmaGauge gauge = new EwmaGauge(0.5);
+
+ gauge.addValue(0L); // seed
+
+ // Step 1: transit=10, prev=0, D=10, J = 0 + (10-0) * 0.5 =
5.0
+ gauge.addValue(10L);
+ assertEquals(5.0, gauge.getValue(), DELTA, "Step 1");
+
+ // Step 2: transit=0, prev=10, D=10, J = 5.0 + (10-5.0) * 0.5 =
7.5
+ gauge.addValue(0L);
+ assertEquals(7.5, gauge.getValue(), DELTA, "Step 2");
+
+ // Step 3: transit=10, prev=0, D=10, J = 7.5 + (10-7.5) * 0.5 =
8.75
+ gauge.addValue(10L);
+ assertEquals(8.75, gauge.getValue(), DELTA, "Step 3");
+ }
+
+ @Test
+ @DisplayName("Zero deviation decays jitter toward zero")
+ void zeroDeviationDecays() {
+ EwmaGauge gauge = new EwmaGauge(0.5);
+ gauge.addValue(0L); // 0
+ gauge.addValue(10L); // 0 + 5*alpha = 2.5
+ double afterFirst = gauge.getValue();
+ assertEquals(afterFirst, gauge.getValue(), DELTA);
+
+ gauge.addValue(10L); // 2.5 - 2.5*alpha = 2.5 - 1.25 = 1.25
+ assertEquals(afterFirst * 0.5, gauge.getValue(), DELTA);
+ }
+
+ }
+
+
+ @Nested
+ @DisplayName("Negative value guard")
+ class NegativeValueTest {
+
+ @Test
+ @DisplayName("Negative transit values are silently ignored before
seed")
+ void negativeIgnoredBeforeSeed() {
+ EwmaGauge gauge = new EwmaGauge();
+ gauge.addValue(-1L);
+ gauge.addValue(-100L);
+ assertEquals(0.0, gauge.getValue(), DELTA);
+ }
+
+ @Test
+ @DisplayName("Negative value after seed does not corrupt lastTransit")
+ void negativeAfterSeedIgnored() {
+ EwmaGauge gauge = new EwmaGauge(0.5);
+ gauge.addValue(10L);
+ gauge.addValue(-5L);
+ gauge.addValue(20L);
+ assertEquals(5.0, gauge.getValue(), DELTA);
+ }
+ }
+
+
+ @Nested
+ @DisplayName("getValue() preserves EWMA across calls")
+ class GetValueIdempotentTest {
+
+ @Test
+ @DisplayName("Repeated getValue() without new samples returns same
estimate")
+ void repeatedGetValueStable() {
+ EwmaGauge gauge = new EwmaGauge(0.5);
+ gauge.addValue(0L);
+ gauge.addValue(10L);
+ double first = gauge.getValue();
+
+ assertEquals(first, gauge.getValue(), DELTA, "Second call");
+ assertEquals(first, gauge.getValue(), DELTA, "Third call");
+ }
+
+ @Test
+ @DisplayName("EWMA accumulates correctly across multiple reporting
windows")
+ void acrossReportingWindows() {
+ EwmaGauge gauge = new EwmaGauge(0.5);
+ gauge.addValue(0L);
+
+ gauge.addValue(10L);
+ assertEquals(5.0, gauge.getValue(), DELTA, "Window 1");
+
+ gauge.addValue(0L);
+ assertEquals(7.5, gauge.getValue(), DELTA, "Window 2");
+
+ gauge.addValue(10L);
+ assertEquals(8.75, gauge.getValue(), DELTA, "Window 3");
+ }
+ }
+
+ @Nested
+ @DisplayName("thread safe")
+ class ConcurrencyTest {
+
+ @Test
+ @DisplayName("Concurrent addValue() calls do not corrupt state")
+ void concurrentAddValue() throws InterruptedException {
+ EwmaGauge gauge = new EwmaGauge();
+ int threads = 8;
+ int samplesPerThread = 10_000;
+ CountDownLatch ready = new CountDownLatch(threads);
+ CountDownLatch start = new CountDownLatch(1);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+
+ for (int t = 0; t < threads; t++) {
+ final long base = t * 10L;
+ pool.submit(() -> {
+ ready.countDown();
+ try {
+ start.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ for (int i = 0; i < samplesPerThread; i++) {
+ gauge.addValue(base + (i % 10));
+ }
+ });
+ }
+
+ ready.await();
+ start.countDown();
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(10, TimeUnit.SECONDS),
+ "Executor did not terminate — possible deadlock");
+
+ double value = gauge.getValue();
+ assertTrue(value >= 0.0, "Jitter must be non-negative, got: " +
value);
+ assertTrue(Double.isFinite(value), "Jitter must be finite, got: "
+ value);
+ }
+
+ @Test
+ @DisplayName("Concurrent getValue() and addValue() do not deadlock")
+ void concurrentGetAndAdd() throws Exception {
+ EwmaGauge gauge = new EwmaGauge();
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ CountDownLatch done = new CountDownLatch(2);
+
+ Future<?> writer = pool.submit(() -> {
+ for (int i = 0; i < 50_000; i++) {
+ gauge.addValue(i % 100);
+ }
+ done.countDown();
+ });
+
+ Future<?> reader = pool.submit(() -> {
+ for (int i = 0; i < 1_000; i++) {
+ double v = gauge.getValue();
+ assertTrue(v >= 0.0 && Double.isFinite(v),
+ "getValue() returned invalid result: " + v);
+ }
+ done.countDown();
+ });
+
+ assertTrue(done.await(10, TimeUnit.SECONDS),
+ "Test did not complete within timeout possible deadlock");
+
+ writer.get();
+ reader.get();
+ pool.shutdown();
+ }
+
+ @Test
+ @DisplayName("Only one thread seeds lastTransit all same value gives
zero jitter")
+ void seedRace() throws InterruptedException {
+ EwmaGauge gauge = new EwmaGauge();
+ int threads = 16;
+ CountDownLatch ready = new CountDownLatch(threads);
+ CountDownLatch start = new CountDownLatch(1);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+
+ for (int t = 0; t < threads; t++) {
+ pool.submit(() -> {
+ ready.countDown();
+ try {
+ start.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ gauge.addValue(42L);
+ });
+ }
+
+ ready.await();
+ start.countDown();
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(5, TimeUnit.SECONDS),
+ "Executor did not terminate possible deadlock");
+
+ assertEquals(0.0, gauge.getValue(), DELTA);
+ }
+ }
+
+ @Nested
+ @DisplayName("Edge cases")
+ class EdgeCaseTest {
+
+ @Test
+ @DisplayName("Long.MAX_VALUE transit does not overflow deviation")
+ void maxLongTransit() {
+ EwmaGauge gauge = new EwmaGauge(0.5);
+ gauge.addValue(0L);
+ gauge.addValue(Long.MAX_VALUE);
+ double value = gauge.getValue();
+ assertTrue(value > 0.0, "Jitter should be positive");
+ assertTrue(Double.isFinite(value), "Jitter should be finite");
+ }
+
+ @Test
+ @DisplayName("Zero transit time is valid and produces zero deviation")
+ void zeroTransit() {
+ EwmaGauge gauge = new EwmaGauge(0.5);
+ gauge.addValue(0L);
+ gauge.addValue(0L);
+ assertEquals(0.0, gauge.getValue(), DELTA);
+ }
+
+ @Test
+ @DisplayName("Large number of samples does not overflow LongAdder")
+ void manySamples() {
+ EwmaGauge gauge = new EwmaGauge();
+ gauge.addValue(0L);
+ for (int i = 1; i <= 100_000; i++) {
+ gauge.addValue(i % 2 == 0 ? 0L : 10L);
+ }
+ double value = gauge.getValue();
+ assertTrue(value > 0.0, "Jitter should be positive after many
samples");
+ assertTrue(value <= 10.0, "Jitter cannot exceed max deviation of
10");
+ assertTrue(Double.isFinite(value), "Jitter must be finite");
+ }
+ }
+}
diff --git
a/storm-client/test/jvm/org/apache/storm/metrics2/TaskMetricsTest.java
b/storm-client/test/jvm/org/apache/storm/metrics2/TaskMetricsTest.java
new file mode 100644
index 000000000..c81299428
--- /dev/null
+++ b/storm-client/test/jvm/org/apache/storm/metrics2/TaskMetricsTest.java
@@ -0,0 +1,485 @@
+/**
+ * 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.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;
+
+ private Map<String, Object> topoConf;
+
+ private TaskMetrics buildTaskMetrics(boolean ewmaEnabled) {
+ try (MockedStatic<ConfigUtils> cfgUtils =
mockStatic(ConfigUtils.class)) {
+ cfgUtils.when(() ->
ConfigUtils.samplingRate(topoConf)).thenReturn(SAMPLING_RATE);
+ cfgUtils.when(() ->
ConfigUtils.ewmaSmoothingFactor(topoConf)).thenReturn(EWMA_FACTOR);
+ cfgUtils.when(() ->
ConfigUtils.ewmaEnable(topoConf)).thenReturn(ewmaEnabled);
+
+ return new TaskMetrics(context, COMPONENT_ID, TASK_ID,
metricRegistry, topoConf);
+ }
+ }
+
+ @BeforeEach
+ void setUp() {
+ when(context.getStormId()).thenReturn(TOPOLOGY_ID);
+ when(context.getThisWorkerPort()).thenReturn(WORKER_PORT);
+
+ topoConf = new HashMap<>();
+
+ when(metricRegistry.rateCounter(anyString(), anyString(), anyString(),
+ anyInt(), anyInt(), anyString())).thenReturn(rateCounter);
+ }
+
+ @Test
+ void spoutAckedTuple_incrementsAckCounter() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.spoutAckedTuple(STREAM_ID, 100L);
+
+ verify(rateCounter).inc(SAMPLING_RATE);
+ }
+
+ @Test
+ void spoutAckedTuple_registersCompleteLatencyGauge() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.spoutAckedTuple(STREAM_ID, 200L);
+
+ verify(metricRegistry, atLeastOnce()).gauge(
+ contains("__complete-latency"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void spoutAckedTuple_withEwmaEnabled_registersJitterGauge() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(true);
+
+ tm.spoutAckedTuple(STREAM_ID, 150L);
+
+ verify(metricRegistry, atLeastOnce()).gauge(
+ contains("__complete-jitter"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void spoutAckedTuple_withEwmaDisabled_doesNotRegisterJitterGauge() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.spoutAckedTuple(STREAM_ID, 150L);
+
+ verify(metricRegistry, never()).gauge(
+ contains("__complete-rfc1889a-jitter"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void boltAckedTuple_incrementsAckCounter() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.boltAckedTuple(SOURCE_COMP, STREAM_ID, 50L);
+
+ verify(rateCounter).inc(SAMPLING_RATE);
+ }
+
+ @Test
+ void boltAckedTuple_registersProcessLatencyGauge() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.boltAckedTuple(SOURCE_COMP, STREAM_ID, 50L);
+
+ verify(metricRegistry, atLeastOnce()).gauge(
+ contains("__process-latency"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void boltAckedTuple_withEwmaEnabled_registersJitterGauge() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(true);
+
+ tm.boltAckedTuple(SOURCE_COMP, STREAM_ID, 75L);
+
+ verify(metricRegistry, atLeastOnce()).gauge(
+ contains("__process-jitter"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void boltAckedTuple_metricKeyIncludesSourceComponentAndStream() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.boltAckedTuple(SOURCE_COMP, STREAM_ID, 50L);
+
+ verify(metricRegistry).rateCounter(
+ contains(SOURCE_COMP + ":" + STREAM_ID),
+ anyString(), anyString(), anyInt(), anyInt(), anyString());
+ }
+
+ @Test
+ void spoutFailedTuple_incrementsFailCounter() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.spoutFailedTuple(STREAM_ID);
+
+ verify(rateCounter).inc(SAMPLING_RATE);
+ }
+
+ @Test
+ void spoutFailedTuple_usesCorrectMetricName() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.spoutFailedTuple(STREAM_ID);
+
+ verify(metricRegistry).rateCounter(
+ eq("__fail-count-" + STREAM_ID),
+ eq(TOPOLOGY_ID), eq(COMPONENT_ID), eq(TASK_ID),
eq(WORKER_PORT), eq(STREAM_ID));
+ }
+
+ @Test
+ void boltFailedTuple_incrementsFailCounter() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.boltFailedTuple(SOURCE_COMP, STREAM_ID);
+
+ verify(rateCounter).inc(SAMPLING_RATE);
+ }
+
+ @Test
+ void boltFailedTuple_metricKeyIncludesSourceComponentAndStream() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.boltFailedTuple(SOURCE_COMP, STREAM_ID);
+
+ verify(metricRegistry).rateCounter(
+ eq("__fail-count-" + SOURCE_COMP + ":" + STREAM_ID),
+ anyString(), anyString(), anyInt(), anyInt(), anyString());
+ }
+
+ @Test
+ void emittedTuple_incrementsEmitCounter() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.emittedTuple(STREAM_ID);
+
+ verify(rateCounter).inc(SAMPLING_RATE);
+ }
+
+ @Test
+ void emittedTuple_usesCorrectMetricName() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.emittedTuple(STREAM_ID);
+
+ verify(metricRegistry).rateCounter(
+ eq("__emit-count-" + STREAM_ID),
+ eq(TOPOLOGY_ID), eq(COMPONENT_ID), eq(TASK_ID),
eq(WORKER_PORT), eq(STREAM_ID));
+ }
+
+ @Test
+ void transferredTuples_incrementsByAmountTimesSamplingRate() {
+ TaskMetrics tm = buildTaskMetrics(false);
+ int amount = 5;
+
+ tm.transferredTuples(STREAM_ID, amount);
+
+ verify(rateCounter).inc(amount * SAMPLING_RATE);
+ }
+
+ @Test
+ void transferredTuples_usesCorrectMetricName() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.transferredTuples(STREAM_ID, 3);
+
+ verify(metricRegistry).rateCounter(
+ eq("__transfer-count-" + STREAM_ID),
+ anyString(), anyString(), anyInt(), anyInt(), anyString());
+ }
+
+ @Test
+ void boltExecuteTuple_incrementsExecuteCounter() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.boltExecuteTuple(SOURCE_COMP, STREAM_ID, 30L);
+
+ verify(rateCounter).inc(SAMPLING_RATE);
+ }
+
+ @Test
+ void boltExecuteTuple_registersExecuteLatencyGauge() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.boltExecuteTuple(SOURCE_COMP, STREAM_ID, 30L);
+
+ verify(metricRegistry, atLeastOnce()).gauge(
+ contains("__execute-latency"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void boltExecuteTuple_withEwmaEnabled_registersJitterGauge() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(true);
+
+ tm.boltExecuteTuple(SOURCE_COMP, STREAM_ID, 30L);
+
+ verify(metricRegistry, atLeastOnce()).gauge(
+ contains("__execute-jitter"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void boltExecuteTuple_withEwmaDisabled_doesNotRegisterJitterGauge() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.boltExecuteTuple(SOURCE_COMP, STREAM_ID, 30L);
+
+ verify(metricRegistry, never()).gauge(
+ contains("__execute-rfc1889a-jitter"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void differentStreams_produceSeparateRateCounters() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.emittedTuple("stream-A");
+ tm.emittedTuple("stream-B");
+
+ verify(metricRegistry).rateCounter(
+ eq("__emit-count-stream-A"),
+ anyString(), anyString(), anyInt(), anyInt(), eq("stream-A"));
+ verify(metricRegistry).rateCounter(
+ eq("__emit-count-stream-B"),
+ anyString(), anyString(), anyInt(), anyInt(), eq("stream-B"));
+ }
+
+ @Test
+ void rateCounter_registeredOnlyOnceForSameMetricName() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.emittedTuple(STREAM_ID);
+ tm.emittedTuple(STREAM_ID);
+ tm.emittedTuple(STREAM_ID);
+
+ verify(metricRegistry, times(1)).rateCounter(
+ eq("__emit-count-" + STREAM_ID),
+ anyString(), anyString(), anyInt(), anyInt(), anyString());
+ }
+
+ @Test
+ void gauge_registeredOnlyOnceForSameMetricName() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.spoutAckedTuple(STREAM_ID, 10L);
+ tm.spoutAckedTuple(STREAM_ID, 20L);
+ tm.spoutAckedTuple(STREAM_ID, 30L);
+
+ verify(metricRegistry, times(1)).gauge(
+ contains("__complete-latency"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void concurrentEmittedTuple_registersRateCounterExactlyOnce() throws
InterruptedException {
+ TaskMetrics tm = buildTaskMetrics(false);
+ int threadCount = 20;
+ CountDownLatch ready = new CountDownLatch(threadCount);
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(threadCount);
+
+ ExecutorService pool = Executors.newFixedThreadPool(threadCount);
+ for (int i = 0; i < threadCount; i++) {
+ pool.submit(() -> {
+ ready.countDown();
+ try { start.await(); } catch (InterruptedException ignored) {}
+ tm.emittedTuple(STREAM_ID);
+ done.countDown();
+ });
+ }
+
+ ready.await();
+ start.countDown();
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ pool.shutdown();
+
+ verify(metricRegistry, times(1)).rateCounter(
+ eq("__emit-count-" + STREAM_ID),
+ anyString(), anyString(), anyInt(), anyInt(), anyString());
+ }
+
+ @Test
+ void concurrentSpoutAckedTuple_registersGaugeExactlyOnce() throws
InterruptedException {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+ int threadCount = 20;
+ CountDownLatch ready = new CountDownLatch(threadCount);
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(threadCount);
+
+ ExecutorService pool = Executors.newFixedThreadPool(threadCount);
+ for (int i = 0; i < threadCount; i++) {
+ pool.submit(() -> {
+ ready.countDown();
+ try { start.await(); } catch (InterruptedException ignored) {}
+ tm.spoutAckedTuple(STREAM_ID, 100L);
+ done.countDown();
+ });
+ }
+
+ ready.await();
+ start.countDown();
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ pool.shutdown();
+
+ verify(metricRegistry, times(1)).gauge(
+ contains("__complete-latency"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void getOrCreateGauge_sameTypeReusedWithoutThrowing() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.spoutAckedTuple(STREAM_ID, 10L);
+ tm.spoutAckedTuple(STREAM_ID, 20L);
+
+ verify(metricRegistry, times(1)).gauge(
+ contains("__complete-latency"), any(Gauge.class),
+ anyString(), anyString(), anyString(), anyInt(), anyInt());
+ }
+
+ @Test
+ void boltAckedTuple_metricNameFormat() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+ String expectedKey = SOURCE_COMP + ":" + STREAM_ID;
+
+ tm.boltAckedTuple(SOURCE_COMP, STREAM_ID, 10L);
+
+ verify(metricRegistry).rateCounter(
+ eq("__ack-count-" + expectedKey),
+ eq(TOPOLOGY_ID), eq(COMPONENT_ID), eq(TASK_ID),
eq(WORKER_PORT), eq(STREAM_ID));
+ }
+
+ @Test
+ void boltExecuteTuple_metricNameFormat() {
+ when(metricRegistry.gauge(anyString(), any(Gauge.class), anyString(),
+ anyString(), anyString(), anyInt(),
anyInt())).thenReturn(null);
+ TaskMetrics tm = buildTaskMetrics(false);
+ String expectedKey = SOURCE_COMP + ":" + STREAM_ID;
+
+ tm.boltExecuteTuple(SOURCE_COMP, STREAM_ID, 10L);
+
+ verify(metricRegistry).rateCounter(
+ eq("__execute-count-" + expectedKey),
+ eq(TOPOLOGY_ID), eq(COMPONENT_ID), eq(TASK_ID),
eq(WORKER_PORT), eq(STREAM_ID));
+ }
+
+ @Test
+ void boltFailedTuple_metricNameFormat() {
+ TaskMetrics tm = buildTaskMetrics(false);
+ String expectedKey = SOURCE_COMP + ":" + STREAM_ID;
+
+ tm.boltFailedTuple(SOURCE_COMP, STREAM_ID);
+
+ verify(metricRegistry).rateCounter(
+ eq("__fail-count-" + expectedKey),
+ eq(TOPOLOGY_ID), eq(COMPONENT_ID), eq(TASK_ID),
eq(WORKER_PORT), eq(STREAM_ID));
+ }
+
+ @Test
+ void contextFields_propagatedCorrectlyToRegistry() {
+ TaskMetrics tm = buildTaskMetrics(false);
+
+ tm.emittedTuple(STREAM_ID);
+
+ ArgumentCaptor<String> topoCaptor =
ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor<String> compCaptor =
ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor<Integer> taskCaptor =
ArgumentCaptor.forClass(Integer.class);
+ ArgumentCaptor<Integer> portCaptor =
ArgumentCaptor.forClass(Integer.class);
+
+ verify(metricRegistry).rateCounter(
+ anyString(),
+ topoCaptor.capture(), compCaptor.capture(),
+ taskCaptor.capture(), portCaptor.capture(),
+ anyString());
+
+ assertEquals(TOPOLOGY_ID, topoCaptor.getValue());
+ assertEquals(COMPONENT_ID, compCaptor.getValue());
+ assertEquals(TASK_ID, taskCaptor.getValue());
+ assertEquals(WORKER_PORT, portCaptor.getValue());
+ }
+}
