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 512b7382c Checkstyle auto-fix of examples/storm-perf (#9176)
512b7382c is described below
commit 512b7382cb2a87319a51f7f257c37353e49916a8
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Oct 18 20:14:23 2026 +0200
Checkstyle auto-fix of examples/storm-perf (#9176)
Co-authored-by: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
---
.../org/apache/storm/perf/BackPressureTopo.java | 10 +--
.../storm/perf/ConstSpoutIdBoltNullBoltTopo.java | 15 ++--
.../apache/storm/perf/ConstSpoutNullBoltTopo.java | 13 ++--
.../org/apache/storm/perf/ConstSpoutOnlyTopo.java | 1 -
.../FileReadWordCountSpoutCompressionTopo.java | 10 ++-
.../apache/storm/perf/FileReadWordCountTopo.java | 10 ++-
.../apache/storm/perf/HdfsSpoutNullBoltTopo.java | 21 +++---
.../storm/perf/JitterAwareGroupingTopology.java | 87 ++++++++++++++--------
.../org/apache/storm/perf/KafkaClientHdfsTopo.java | 41 +++++-----
.../storm/perf/KafkaClientSpoutNullBoltTopo.java | 33 ++++----
.../org/apache/storm/perf/LowThroughputTopo.java | 7 +-
.../apache/storm/perf/SimplifiedWordCountTopo.java | 7 +-
.../apache/storm/perf/StrGenSpoutHdfsBoltTopo.java | 29 ++++----
.../org/apache/storm/perf/ThroughputMeter.java | 1 +
.../java/org/apache/storm/perf/bolt/CountBolt.java | 1 -
.../org/apache/storm/perf/bolt/DevNullBolt.java | 4 +-
.../java/org/apache/storm/perf/bolt/IdBolt.java | 3 +-
.../apache/storm/perf/bolt/SplitSentenceBolt.java | 1 -
.../org/apache/storm/perf/queuetest/Acker.java | 1 -
.../storm/perf/queuetest/JCQueuePerfTest.java | 29 +++++---
.../org/apache/storm/perf/spout/ConstSpout.java | 3 +-
.../org/apache/storm/perf/spout/FileReadSpout.java | 1 -
.../apache/storm/perf/spout/StringGenSpout.java | 7 +-
.../org/apache/storm/perf/spout/WordGenSpout.java | 1 -
.../storm/perf/toolstest/JCToolsPerfTest.java | 4 +-
.../storm/perf/utils/BasicMetricsCollector.java | 31 ++++----
.../java/org/apache/storm/perf/utils/Helper.java | 14 ++--
.../org/apache/storm/perf/utils/IdentityBolt.java | 3 +-
.../org/apache/storm/perf/utils/MetricsSample.java | 4 +-
29 files changed, 220 insertions(+), 172 deletions(-)
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/BackPressureTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/BackPressureTopo.java
index 5fb70220e..f2a130213 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/BackPressureTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/BackPressureTopo.java
@@ -34,7 +34,6 @@ import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Utils;
import org.slf4j.LoggerFactory;
-
public class BackPressureTopo {
private static final String SPOUT_ID = "ConstSpout";
@@ -52,7 +51,6 @@ public class BackPressureTopo {
// 2 - Setup DevNull Bolt --------
ThrottledBolt bolt = new ThrottledBolt(sleepMs);
-
// 3 - Setup Topology --------
TopologyBuilder builder = new TopologyBuilder();
@@ -80,7 +78,8 @@ public class BackPressureTopo {
return;
}
// Submit topology to storm cluster
- Helper.runOnClusterAndPrintMetrics(runTime, "BackPressureTopo",
topoConf, getTopology(topoConf));
+ Helper.runOnClusterAndPrintMetrics(runTime, "BackPressureTopo",
topoConf,
+ getTopology(topoConf));
}
private static class ThrottledBolt extends BaseRichBolt {
@@ -93,7 +92,8 @@ public class BackPressureTopo {
}
@Override
- public void prepare(Map<String, Object> topoConf, TopologyContext
context, OutputCollector collector) {
+ public void prepare(Map<String, Object> topoConf, TopologyContext
context,
+ OutputCollector collector) {
this.collector = collector;
}
@@ -104,7 +104,7 @@ public class BackPressureTopo {
try {
Thread.sleep(sleepMs);
} catch (InterruptedException e) {
- //.. ignore
+ // .. ignore
}
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutIdBoltNullBoltTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutIdBoltNullBoltTopo.java
index c83763d6e..514a4f3b7 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutIdBoltNullBoltTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutIdBoltNullBoltTopo.java
@@ -29,8 +29,10 @@ import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
/**
- * ConstSpout -> IdBolt -> DevNullBolt This topology measures speed of
messaging between spouts->bolt and bolt->bolt ConstSpout :
- * Continuously emits a constant string IdBolt : clones and emits input tuples
DevNullBolt : discards incoming tuples.
+ * ConstSpout -> IdBolt -> DevNullBolt This topology measures speed of
messaging between
+ * spouts->bolt and bolt->bolt ConstSpout :
+ * Continuously emits a constant string IdBolt : clones and emits input tuples
DevNullBolt :
+ * discards incoming tuples.
*/
public class ConstSpoutIdBoltNullBoltTopo {
@@ -53,7 +55,6 @@ public class ConstSpoutIdBoltNullBoltTopo {
IdBolt bolt1 = new IdBolt();
DevNullBolt bolt2 = new DevNullBolt();
-
// 3 - Setup Topology --------
TopologyBuilder builder = new TopologyBuilder();
@@ -62,16 +63,16 @@ public class ConstSpoutIdBoltNullBoltTopo {
int numBolt1 = Helper.getInt(conf, BOLT1_COUNT, 1);
builder.setBolt(BOLT1_ID, bolt1, numBolt1)
- .localOrShuffleGrouping(SPOUT_ID);
+ .localOrShuffleGrouping(SPOUT_ID);
int numBolt2 = Helper.getInt(conf, BOLT2_COUNT, 1);
builder.setBolt(BOLT2_ID, bolt2, numBolt2)
- .localOrShuffleGrouping(BOLT1_ID);
- System.err.printf("====> Using : numSpouts = %d , numBolt1 = %d,
numBolt2=%d\n", numSpouts, numBolt1, numBolt2);
+ .localOrShuffleGrouping(BOLT1_ID);
+ System.err.printf("====> Using : numSpouts = %d , numBolt1 = %d,
numBolt2=%d\n", numSpouts,
+ numBolt1, numBolt2);
return builder.createTopology();
}
-
public static void main(String[] args) throws Exception {
int runTime = -1;
Config topoConf = new Config();
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutNullBoltTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutNullBoltTopo.java
index 63ef51b45..be0da3c7a 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutNullBoltTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutNullBoltTopo.java
@@ -58,7 +58,6 @@ public class ConstSpoutNullBoltTopo {
// 2 - Setup DevNull Bolt --------
DevNullBolt bolt = new DevNullBolt();
-
// 3 - Setup Topology --------
TopologyBuilder builder = new TopologyBuilder();
@@ -87,11 +86,15 @@ public class ConstSpoutNullBoltTopo {
Config topoConf = new Config();
// Configured for achieving max throughput in single worker mode
(empirically found).
// For reference : numbers taken on MacBook Pro mid 2015
- // -- ACKer=0: ~8 mill/sec (batchSz=2k & recvQsize=50k). 6.7
mill/sec (batchSz=1 & recvQsize=1k)
- // -- ACKer=1: ~1 mill/sec, lat= ~1 microsec (batchSz=1 &
bolt.wait.strategy=Park bolt.wait.park.micros=0)
- // -- ACKer=1: ~1.3 mill/sec, lat= ~11 micros (batchSz=1 &
receive.buffer.size=1k, bolt.wait & bp.wait =
+ // -- ACKer=0: ~8 mill/sec (batchSz=2k & recvQsize=50k). 6.7 mill/sec
(batchSz=1 &
+ // recvQsize=1k)
+ // -- ACKer=1: ~1 mill/sec, lat= ~1 microsec (batchSz=1 &
bolt.wait.strategy=Park
+ // bolt.wait.park.micros=0)
+ // -- ACKer=1: ~1.3 mill/sec, lat= ~11 micros (batchSz=1 &
receive.buffer.size=1k, bolt.wait
+ // & bp.wait =
// Progressive[defaults])
- // -- ACKer=1: ~1.6 mill/sec, lat= ~300 micros (batchSz=500 &
bolt.wait.strategy=Park bolt.wait.park.micros=0)
+ // -- ACKer=1: ~1.6 mill/sec, lat= ~300 micros (batchSz=500 &
bolt.wait.strategy=Park
+ // bolt.wait.park.micros=0)
topoConf.put(Config.TOPOLOGY_SPOUT_RECVQ_SKIPS, 8);
topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 500);
topoConf.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, 50_000);
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutOnlyTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutOnlyTopo.java
index 1b012fe2f..44eb2cb37 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutOnlyTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ConstSpoutOnlyTopo.java
@@ -35,7 +35,6 @@ public class ConstSpoutOnlyTopo {
public static final String TOPOLOGY_NAME = "ConstSpoutOnlyTopo";
public static final String SPOUT_ID = "constSpout";
-
static StormTopology getTopology() {
// 1 - Setup Const Spout --------
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountSpoutCompressionTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountSpoutCompressionTopo.java
index 51c43057e..a15aa0f7f 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountSpoutCompressionTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountSpoutCompressionTopo.java
@@ -50,7 +50,6 @@ public class FileReadWordCountSpoutCompressionTopo {
public static final int DEFAULT_SPLIT_BOLT_NUM = 2;
public static final int DEFAULT_COUNT_BOLT_NUM = 2;
-
static StormTopology getTopology(Map<String, Object> config) {
final int spoutNum = Helper.getInt(config, SPOUT_NUM,
DEFAULT_SPOUT_NUM);
@@ -62,8 +61,10 @@ public class FileReadWordCountSpoutCompressionTopo {
// sampledata/longrandomwords.txt contains sentences with at least
1500 bytes
builder.setSpout(SPOUT_ID, new FileReadSpout(inputFile), spoutNum)
.addConfiguration(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE,
true);
- builder.setBolt(SPLIT_ID, new SplitSentenceBolt(),
spBoltNum).localOrShuffleGrouping(SPOUT_ID);
- builder.setBolt(COUNT_ID, new CountBolt(),
cntBoltNum).fieldsGrouping(SPLIT_ID, new Fields(SplitSentenceBolt.FIELDS));
+ builder.setBolt(SPLIT_ID, new SplitSentenceBolt(), spBoltNum)
+ .localOrShuffleGrouping(SPOUT_ID);
+ builder.setBolt(COUNT_ID, new CountBolt(),
cntBoltNum).fieldsGrouping(SPLIT_ID,
+ new Fields(SplitSentenceBolt.FIELDS));
return builder.createTopology();
}
@@ -78,7 +79,8 @@ public class FileReadWordCountSpoutCompressionTopo {
topoConf.putAll(Utils.findAndReadConfigFile(args[1]));
}
topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
"org.apache.storm.policy.WaitStrategyPark");
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
+ "org.apache.storm.policy.WaitStrategyPark");
topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountTopo.java
index 78eaab7ef..a0f9c5f8e 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountTopo.java
@@ -50,7 +50,6 @@ public class FileReadWordCountTopo {
public static final int DEFAULT_SPLIT_BOLT_NUM = 2;
public static final int DEFAULT_COUNT_BOLT_NUM = 2;
-
static StormTopology getTopology(Map<String, Object> config) {
final int spoutNum = Helper.getInt(config, SPOUT_NUM,
DEFAULT_SPOUT_NUM);
@@ -60,8 +59,10 @@ public class FileReadWordCountTopo {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT_ID, new FileReadSpout(inputFile), spoutNum);
- builder.setBolt(SPLIT_ID, new SplitSentenceBolt(),
spBoltNum).localOrShuffleGrouping(SPOUT_ID);
- builder.setBolt(COUNT_ID, new CountBolt(),
cntBoltNum).fieldsGrouping(SPLIT_ID, new Fields(SplitSentenceBolt.FIELDS));
+ builder.setBolt(SPLIT_ID, new SplitSentenceBolt(), spBoltNum)
+ .localOrShuffleGrouping(SPOUT_ID);
+ builder.setBolt(COUNT_ID, new CountBolt(),
cntBoltNum).fieldsGrouping(SPLIT_ID,
+ new Fields(SplitSentenceBolt.FIELDS));
return builder.createTopology();
}
@@ -76,7 +77,8 @@ public class FileReadWordCountTopo {
topoConf.putAll(Utils.findAndReadConfigFile(args[1]));
}
topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
"org.apache.storm.policy.WaitStrategyPark");
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
+ "org.apache.storm.policy.WaitStrategyPark");
topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/HdfsSpoutNullBoltTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/HdfsSpoutNullBoltTopo.java
index 9876cac9f..d3be4d9d2 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/HdfsSpoutNullBoltTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/HdfsSpoutNullBoltTopo.java
@@ -60,15 +60,14 @@ public class HdfsSpoutNullBoltTopo {
final String archiveDir = Helper.getStr(config, ARCHIVE_DIR);
final String badDir = Helper.getStr(config, BAD_DIR);
-
// 1 - Setup Hdfs Spout --------
HdfsSpout spout = new HdfsSpout()
- .setReaderType(fileFormat)
- .setHdfsUri(hdfsUri)
- .setSourceDir(sourceDir)
- .setArchiveDir(archiveDir)
- .setBadFilesDir(badDir)
- .withOutputFields(TextFileReader.defaultFields);
+ .setReaderType(fileFormat)
+ .setHdfsUri(hdfsUri)
+ .setSourceDir(sourceDir)
+ .setArchiveDir(archiveDir)
+ .setBadFilesDir(badDir)
+ .withOutputFields(TextFileReader.defaultFields);
// 2 - DevNull Bolt --------
DevNullBolt bolt = new DevNullBolt();
@@ -77,7 +76,7 @@ public class HdfsSpoutNullBoltTopo {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT_ID, spout, spoutNum);
builder.setBolt(BOLT_ID, bolt, boltNum)
- .localOrShuffleGrouping(SPOUT_ID);
+ .localOrShuffleGrouping(SPOUT_ID);
return builder.createTopology();
}
@@ -92,13 +91,15 @@ public class HdfsSpoutNullBoltTopo {
Config topoConf = new Config();
topoConf.putAll(Utils.findAndReadConfigFile(args[1]));
topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
"org.apache.storm.policy.WaitStrategyPark");
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
+ "org.apache.storm.policy.WaitStrategyPark");
topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
topoConf.putAll(Utils.readCommandLineOpts());
// Submit to Storm cluster
- Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME,
topoConf, getTopology(topoConf));
+ Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME,
topoConf,
+ getTopology(topoConf));
}
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/JitterAwareGroupingTopology.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/JitterAwareGroupingTopology.java
index 7bc53476f..d1a722d39 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/JitterAwareGroupingTopology.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/JitterAwareGroupingTopology.java
@@ -52,14 +52,17 @@ import org.apache.storm.utils.Utils;
*
* <p>{@code JitteryWorkerBolt} tasks have task-index-dependent latency
<i>jitter</i>: task 0 is
* perfectly steady and each higher index is progressively jitterier, because
the per-tuple noise
- * <i>width</i> grows with the task index over a constant floor. This
deliberately exercises the signal
+ * <i>width</i> grows with the task index over a constant floor. This
deliberately exercises the
+ * signal
* the grouping ranks on — RFC-1889 jitter (the EWMA of {@code |Δlatency|}),
which measures dispersion,
* not level. A per-task <i>mean</i> offset would not work: it cancels in the
estimator's
* consecutive-difference, leaving every task with identical jitter. With
upstream feedback enabled,
- * {@link JitterAwareStreamGrouping} steers more tuples toward the steadiest
(lowest-jitter) tasks; run it
+ * {@link JitterAwareStreamGrouping} steers more tuples toward the steadiest
(lowest-jitter) tasks;
+ * run it
* against the {@code loadaware} baseline mode to measure the effect.
*
- * <p>Run the baseline and the jitter-aware run back-to-back to compare.
Select the grouping with the
+ * <p>Run the baseline and the jitter-aware run back-to-back to compare.
Select the grouping with
+ * the
* {@code grouping.mode} flag:
* <pre>
* # Baseline: plain LoadAwareShuffleGrouping (no upstream feedback needed).
@@ -120,18 +123,18 @@ public class JitterAwareGroupingTopology {
String inputFile = Helper.getStr(conf, INPUT_FILE);
if (inputFile == null || inputFile.trim().isEmpty()) {
throw new IllegalArgumentException(
- "Required config '" + INPUT_FILE + "' is not set. Pass a
sentence file, e.g. "
- + "-c " + INPUT_FILE + "=src/main/sampledata/randomwords.txt");
+ "Required config '" + INPUT_FILE + "' is not set. Pass a
sentence file, e.g. "
+ + "-c " + INPUT_FILE +
"=src/main/sampledata/randomwords.txt");
}
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT_ID, new GenSpout(inputFile), spouts);
builder.setBolt(SPLITTER_ID, new SplitterBolt(), splitters)
- .localOrShuffleGrouping(SPOUT_ID);
+ .localOrShuffleGrouping(SPOUT_ID);
builder.setBolt(WORKER_ID, new JitteryWorkerBolt(baseDelayUs), workers)
- .customGrouping(SPLITTER_ID, selectGrouping(conf));
+ .customGrouping(SPLITTER_ID, selectGrouping(conf));
builder.setBolt(SINK_ID, new SinkBolt(), sinks)
- .localOrShuffleGrouping(WORKER_ID);
+ .localOrShuffleGrouping(WORKER_ID);
return builder.createTopology();
}
@@ -139,8 +142,10 @@ public class JitterAwareGroupingTopology {
/**
* Picks the {@code splitter -> worker} grouping from {@link
#GROUPING_MODE}. Defaults to the
* feedback-driven {@link JitterAwareStreamGrouping}; {@code
grouping.mode=loadaware} selects the plain
- * {@link LoadAwareShuffleGrouping} baseline so the two can be benchmarked
back-to-back. Both implement
- * {@link org.apache.storm.grouping.LoadAwareCustomStreamGrouping} and
require the locality-aware
+ * {@link LoadAwareShuffleGrouping} baseline so the two can be benchmarked
back-to-back. Both
+ * implement
+ * {@link org.apache.storm.grouping.LoadAwareCustomStreamGrouping} and
require the
+ * locality-aware
* configuration set in {@link #main}.
*/
static CustomStreamGrouping selectGrouping(Map<String, Object> conf) {
@@ -166,19 +171,26 @@ public class JitterAwareGroupingTopology {
}
topoConf.put(Config.TOPOLOGY_STATS_EWMA_ENABLE, true);
- // max.spout.pending counts SPOUT tuples (sentences), but each fans
out to ~7 word-tuples at the
- // SplitterBolt, so the in-flight backlog at the slow workers is ~7x
this number. Keep it low enough
- // that steady-state complete latency stays well under
topology.message.timeout.secs with ~0 fails;
- // 4000 oversubscribed the synthetic workers and pinned latency at the
timeout. Tune per cluster:
+ // max.spout.pending counts SPOUT tuples (sentences), but each fans
out to ~7 word-tuples at
+ // the
+ // SplitterBolt, so the in-flight backlog at the slow workers is ~7x
this number. Keep it
+ // low enough
+ // that steady-state complete latency stays well under
topology.message.timeout.secs with ~0
+ // fails;
+ // 4000 oversubscribed the synthetic workers and pinned latency at the
timeout. Tune per
+ // cluster:
// complete_latency ~= (pending * fanout) / aggregate_worker_ack_rate
(Little's law).
topoConf.putIfAbsent(Config.TOPOLOGY_MAX_SPOUT_PENDING, 500);
topoConf.putIfAbsent(Config.TOPOLOGY_UPSTREAM_FEEDBACK_FREQ_SECS, 10);
- // Both grouping modes resolve to a LoadAwareShuffleGrouping
(directly, or as the jitter grouping's
- // fallback), whose prepare() requires these locality-aware keys.
Normally supplied by defaults.yaml;
- // set defensively so the topology is self-contained. Placed before
the CLI merge so -c can override.
+ // Both grouping modes resolve to a LoadAwareShuffleGrouping
(directly, or as the jitter
+ // grouping's
+ // fallback), whose prepare() requires these locality-aware keys.
Normally supplied by
+ // defaults.yaml;
+ // set defensively so the topology is self-contained. Placed before
the CLI merge so -c can
+ // override.
topoConf.putIfAbsent(Config.STORM_NETWORK_TOPOGRAPHY_PLUGIN,
-
"org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping");
+
"org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping");
topoConf.putIfAbsent(Config.TOPOLOGY_LOCALITYAWARE_HIGHER_BOUND, 0.8);
topoConf.putIfAbsent(Config.TOPOLOGY_LOCALITYAWARE_LOWER_BOUND, 0.2);
@@ -205,7 +217,8 @@ public class JitterAwareGroupingTopology {
}
@Override
- public void open(Map<String, Object> conf, TopologyContext context,
SpoutOutputCollector collector) {
+ public void open(Map<String, Object> conf, TopologyContext context,
+ SpoutOutputCollector collector) {
this.collector = collector;
try {
this.lines = FileReadSpout.readLines(new
FileInputStream(filePath));
@@ -237,7 +250,8 @@ public class JitterAwareGroupingTopology {
private OutputCollector collector;
@Override
- public void prepare(Map<String, Object> conf, TopologyContext context,
OutputCollector collector) {
+ public void prepare(Map<String, Object> conf, TopologyContext context,
+ OutputCollector collector) {
this.collector = collector;
}
@@ -257,12 +271,14 @@ public class JitterAwareGroupingTopology {
}
/**
- * Counts words and parks for a constant floor plus
task-index-proportional random noise, so each
+ * Counts words and parks for a constant floor plus
task-index-proportional random noise, so
+ * each
* task's RFC-1889 execute-jitter (the EWMA of {@code |Δ
execute-latency|}) differs by construction.
*
* <p>Task {@code i} parks for {@code baseDelayUs} µs plus uniform noise
in {@code [0, i * baseDelayUs]}
* µs per tuple. Crucially, the constant floor cancels in the
consecutive-difference the jitter
- * estimator takes, so only the noise <i>width</i> drives jitter — not a
per-task mean offset (which
+ * estimator takes, so only the noise <i>width</i> drives jitter — not a
per-task mean offset
+ * (which
* would cancel and leave every task with identical jitter). Steady-state
jitter is {@code width / 3}.
* For a 4-task setup with the default {@code baseDelayUs = 2000}:
* <ul>
@@ -271,11 +287,14 @@ public class JitterAwareGroupingTopology {
* <li>Task 2: ~2–6 ms — jitter ≈ 1.33 ms</li>
* <li>Task 3: ~2–8 ms — jitter ≈ 2 ms (jitteriest)</li>
* </ul>
- * Widths are millisecond-scale so the EWMA jitter gauge (millisecond
resolution) records distinct
- * values per task. {@link JitterAwareStreamGrouping} then steers tuples
toward the lowest-jitter
+ * Widths are millisecond-scale so the EWMA jitter gauge (millisecond
resolution) records
+ * distinct
+ * values per task. {@link JitterAwareStreamGrouping} then steers tuples
toward the
+ * lowest-jitter
* task (task 0) when feedback is enabled.
*
- * <p>{@link LockSupport#parkNanos} is used instead of a spin loop so
jittery tasks yield the CPU
+ * <p>{@link LockSupport#parkNanos} is used instead of a spin loop so
jittery tasks yield the
+ * CPU
* and do not starve the steady task's executor thread.
*/
private static class JitteryWorkerBolt extends BaseRichBolt {
@@ -290,14 +309,18 @@ public class JitterAwareGroupingTopology {
}
@Override
- public void prepare(Map<String, Object> conf, TopologyContext context,
OutputCollector collector) {
+ public void prepare(Map<String, Object> conf, TopologyContext context,
+ OutputCollector collector) {
this.collector = collector;
long baseDelayNs = baseDelayUs * 1_000L;
- // Constant floor parked by every task: keeps each task doing real
work, but cancels in the
+ // Constant floor parked by every task: keeps each task doing real
work, but cancels in
+ // the
// consecutive-difference the RFC-1889 jitter estimator takes, so
it adds no jitter.
this.baseFloorNs = baseDelayNs;
- // Noise WIDTH grows with task index, so execute-jitter (EWMA of
|Δlatency|) genuinely differs
- // per task: task 0 is perfectly steady (zero jitter), higher
indices are progressively jitterier.
+ // Noise WIDTH grows with task index, so execute-jitter (EWMA of
|Δlatency|) genuinely
+ // differs
+ // per task: task 0 is perfectly steady (zero jitter), higher
indices are progressively
+ // jitterier.
this.jitterWidthNs = baseDelayNs * context.getThisTaskIndex();
}
@@ -307,7 +330,8 @@ public class JitterAwareGroupingTopology {
counts.merge(word, 1, Integer::sum);
int count = counts.get(word);
- long noiseNs = jitterWidthNs == 0L ? 0L : (long)
(ThreadLocalRandom.current().nextDouble() * jitterWidthNs);
+ long noiseNs = jitterWidthNs == 0L ? 0L : (long)
(ThreadLocalRandom.current()
+ .nextDouble() * jitterWidthNs);
long sleepNs = baseFloorNs + noiseNs;
if (sleepNs > 0) {
LockSupport.parkNanos(sleepNs);
@@ -332,7 +356,8 @@ public class JitterAwareGroupingTopology {
private OutputCollector collector;
@Override
- public void prepare(Map<String, Object> conf, TopologyContext context,
OutputCollector collector) {
+ public void prepare(Map<String, Object> conf, TopologyContext context,
+ OutputCollector collector) {
this.collector = collector;
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientHdfsTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientHdfsTopo.java
index 5eb61b2e4..8a9a7dd9a 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientHdfsTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientHdfsTopo.java
@@ -58,7 +58,6 @@ public class KafkaClientHdfsTopo {
public static final String HDFS_PATH = "hdfs.dir";
public static final String HDFS_BATCH = "hdfs.batch";
-
public static final int DEFAULT_SPOUT_NUM = 1;
public static final int DEFAULT_BOLT_NUM = 1;
public static final int DEFAULT_HDFS_BATCH = 1000;
@@ -68,7 +67,6 @@ public class KafkaClientHdfsTopo {
public static final String SPOUT_ID = "kafkaSpout";
public static final String BOLT_ID = "hdfsBolt";
-
static StormTopology getTopology(Map<String, Object> config) {
final int spoutNum = getInt(config, SPOUT_NUM, DEFAULT_SPOUT_NUM);
@@ -80,10 +78,11 @@ public class KafkaClientHdfsTopo {
String bootstrapHosts = getStr(config, KAFKA_BOOTSTRAP_HOSTS);
String topicName = getStr(config, KAFKA_TOPIC);
- KafkaSpoutConfig<String, String> spoutConfig =
KafkaSpoutConfig.builder(bootstrapHosts, topicName)
-
.setFirstPollOffsetStrategy(
-
FirstPollOffsetStrategy.EARLIEST)
-
.build();
+ KafkaSpoutConfig<String, String> spoutConfig =
KafkaSpoutConfig.builder(bootstrapHosts,
+ topicName)
+ .setFirstPollOffsetStrategy(
+ FirstPollOffsetStrategy.EARLIEST)
+ .build();
KafkaSpout<String, String> spout = new KafkaSpout<>(spoutConfig);
@@ -91,29 +90,29 @@ public class KafkaClientHdfsTopo {
String hdfsUrls = getStr(config, HDFS_URI);
RecordFormat format = new LineWriter("value");
SyncPolicy syncPolicy = new CountSyncPolicy(hdfsBatch);
- FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(1.0f,
FileSizeRotationPolicy.Units.GB);
+ FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(1.0f,
+ FileSizeRotationPolicy.Units.GB);
- FileNameFormat fileNameFormat = new
DefaultFileNameFormat().withPath(getStr(config, HDFS_PATH));
+ FileNameFormat fileNameFormat = new
DefaultFileNameFormat().withPath(getStr(config,
+ HDFS_PATH));
// Instantiate the HdfsBolt
HdfsBolt bolt = new HdfsBolt()
- .withFsUrl(hdfsUrls)
- .withFileNameFormat(fileNameFormat)
- .withRecordFormat(format)
- .withRotationPolicy(rotationPolicy)
- .withSyncPolicy(syncPolicy);
-
+ .withFsUrl(hdfsUrls)
+ .withFileNameFormat(fileNameFormat)
+ .withRecordFormat(format)
+ .withRotationPolicy(rotationPolicy)
+ .withSyncPolicy(syncPolicy);
// 3 - Setup Topology --------
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT_ID, spout, spoutNum);
builder.setBolt(BOLT_ID, bolt, boltNum)
- .localOrShuffleGrouping(SPOUT_ID);
+ .localOrShuffleGrouping(SPOUT_ID);
return builder.createTopology();
}
-
public static int getInt(Map map, Object key, int def) {
return ObjectReader.getInt(Utils.get(map, key, def));
}
@@ -122,9 +121,9 @@ public class KafkaClientHdfsTopo {
return (String) map.get(key);
}
-
/**
- * Copies text file content from sourceDir to destinationDir. Moves source
files into sourceDir after its done consuming.
+ * Copies text file content from sourceDir to destinationDir. Moves source
files into sourceDir
+ * after its done consuming.
*/
public static void main(String[] args) throws Exception {
@@ -138,13 +137,15 @@ public class KafkaClientHdfsTopo {
topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
"org.apache.storm.policy.WaitStrategyPark");
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
+ "org.apache.storm.policy.WaitStrategyPark");
topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
topoConf.putAll(Utils.readCommandLineOpts());
// Submit topology to Storm cluster
Integer durationSec = Integer.parseInt(args[0]);
- Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME,
topoConf, getTopology(topoConf));
+ Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME,
topoConf,
+ getTopology(topoConf));
}
public static class LineWriter implements RecordFormat {
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientSpoutNullBoltTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientSpoutNullBoltTopo.java
index 4ac1ed8e8..d5d725b73 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientSpoutNullBoltTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/KafkaClientSpoutNullBoltTopo.java
@@ -30,7 +30,8 @@ import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
/**
- * Benchmark topology for measuring spout read/emit/ack performance. The spout
reads and emits tuples. The bolt acks and discards received
+ * Benchmark topology for measuring spout read/emit/ack performance. The spout
reads and emits
+ * tuples. The bolt acks and discards received
* tuples.
*/
public class KafkaClientSpoutNullBoltTopo {
@@ -62,20 +63,23 @@ public class KafkaClientSpoutNullBoltTopo {
final int boltNum = Helper.getInt(config, BOLT_NUM, DEFAULT_BOLT_NUM);
// 1 - Setup Kafka Spout --------
- String bootstrapServers = Optional.ofNullable(Helper.getStr(config,
BOOTSTRAP_SERVERS)).orElse("127.0.0.1:9092");
- String kafkaTopic = Optional.ofNullable(Helper.getStr(config,
KAFKA_TOPIC)).orElse("storm-perf-null-bolt-topic");
+ String bootstrapServers = Optional.ofNullable(Helper.getStr(config,
BOOTSTRAP_SERVERS))
+ .orElse("127.0.0.1:9092");
+ String kafkaTopic = Optional.ofNullable(Helper.getStr(config,
KAFKA_TOPIC))
+ .orElse("storm-perf-null-bolt-topic");
ProcessingGuarantee processingGuarantee = ProcessingGuarantee.valueOf(
- Optional.ofNullable(Helper.getStr(config, PROCESSING_GUARANTEE))
- .orElse(ProcessingGuarantee.AT_LEAST_ONCE.name()));
+ Optional.ofNullable(Helper.getStr(config,
PROCESSING_GUARANTEE))
+ .orElse(ProcessingGuarantee.AT_LEAST_ONCE.name()));
int offsetCommitPeriodMs = Helper.getInt(config,
OFFSET_COMMIT_PERIOD_MS, 30_000);
- KafkaSpoutConfig<String, String> kafkaSpoutConfig =
KafkaSpoutConfig.builder(bootstrapServers, kafkaTopic)
-
.setProcessingGuarantee(processingGuarantee)
-
.setOffsetCommitPeriodMs(offsetCommitPeriodMs)
-
.setFirstPollOffsetStrategy(
-
FirstPollOffsetStrategy.EARLIEST)
-
.setTupleTrackingEnforced(true)
-
.build();
+ KafkaSpoutConfig<String, String> kafkaSpoutConfig = KafkaSpoutConfig
+ .builder(bootstrapServers, kafkaTopic)
+ .setProcessingGuarantee(processingGuarantee)
+ .setOffsetCommitPeriodMs(offsetCommitPeriodMs)
+ .setFirstPollOffsetStrategy(
+ FirstPollOffsetStrategy.EARLIEST)
+ .setTupleTrackingEnforced(true)
+ .build();
KafkaSpout<String, String> spout = new KafkaSpout<>(kafkaSpoutConfig);
@@ -86,7 +90,7 @@ public class KafkaClientSpoutNullBoltTopo {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT_ID, spout, spoutNum);
builder.setBolt(BOLT_ID, bolt, boltNum)
- .localOrShuffleGrouping(SPOUT_ID);
+ .localOrShuffleGrouping(SPOUT_ID);
return builder.createTopology();
}
@@ -109,7 +113,8 @@ public class KafkaClientSpoutNullBoltTopo {
}
// Submit to Storm cluster
- Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME,
topoConf, getTopology(topoConf));
+ Helper.runOnClusterAndPrintMetrics(durationSec, TOPOLOGY_NAME,
topoConf,
+ getTopology(topoConf));
}
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/LowThroughputTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/LowThroughputTopo.java
index 71ffd7cae..c75bbf73d 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/LowThroughputTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/LowThroughputTopo.java
@@ -49,12 +49,12 @@ public class LowThroughputTopo {
Long sleepMs = ObjectReader.getLong(conf.get(SLEEP_MS));
// 1 - Setup Spout --------
- ThrottledSpout spout = new
ThrottledSpout(sleepMs).withOutputFields(ThrottledSpout.DEFAULT_FIELD_NAME);
+ ThrottledSpout spout = new ThrottledSpout(sleepMs)
+ .withOutputFields(ThrottledSpout.DEFAULT_FIELD_NAME);
// 2 - Setup DevNull Bolt --------
LatencyPrintBolt bolt = new LatencyPrintBolt();
-
// 3 - Setup Topology --------
TopologyBuilder builder = new TopologyBuilder();
@@ -83,7 +83,8 @@ public class LowThroughputTopo {
}
topoConf.putAll(Utils.readCommandLineOpts());
// Submit topology to storm cluster
- Helper.runOnClusterAndPrintMetrics(runTime, "LowThroughputTopo",
topoConf, getTopology(topoConf));
+ Helper.runOnClusterAndPrintMetrics(runTime, "LowThroughputTopo",
topoConf,
+ getTopology(topoConf));
}
private static class ThrottledSpout extends BaseRichSpout {
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/SimplifiedWordCountTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/SimplifiedWordCountTopo.java
index 0a9148c59..24d6175c5 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/SimplifiedWordCountTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/SimplifiedWordCountTopo.java
@@ -42,7 +42,6 @@ public class SimplifiedWordCountTopo {
public static final int DEFAULT_SPOUT_NUM = 1;
public static final int DEFAULT_COUNT_BOLT_NUM = 1;
-
static StormTopology getTopology(Map<String, Object> config) {
final int spoutNum = Helper.getInt(config, SPOUT_NUM,
DEFAULT_SPOUT_NUM);
@@ -51,7 +50,8 @@ public class SimplifiedWordCountTopo {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT_ID, new WordGenSpout(inputFile), spoutNum);
- builder.setBolt(COUNT_ID, new CountBolt(),
cntBoltNum).fieldsGrouping(SPOUT_ID, new Fields(WordGenSpout.FIELDS));
+ builder.setBolt(COUNT_ID, new CountBolt(),
cntBoltNum).fieldsGrouping(SPOUT_ID,
+ new Fields(WordGenSpout.FIELDS));
return builder.createTopology();
}
@@ -74,7 +74,8 @@ public class SimplifiedWordCountTopo {
topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
"org.apache.storm.policy.WaitStrategyPark");
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
+ "org.apache.storm.policy.WaitStrategyPark");
topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
topoConf.putAll(Utils.readCommandLineOpts());
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/StrGenSpoutHdfsBoltTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/StrGenSpoutHdfsBoltTopo.java
index 0c53db97c..69d7463f5 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/StrGenSpoutHdfsBoltTopo.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/StrGenSpoutHdfsBoltTopo.java
@@ -16,7 +16,6 @@
* limitations under the License
*/
-
package org.apache.storm.perf;
import java.util.Map;
@@ -63,51 +62,49 @@ public class StrGenSpoutHdfsBoltTopo {
public static final String SPOUT_ID = "GenSpout";
public static final String BOLT_ID = "hdfsBolt";
-
static StormTopology getTopology(Map<String, Object> topoConf) {
final int hdfsBatch = Helper.getInt(topoConf, HDFS_BATCH,
DEFAULT_HDFS_BATCH);
// 1 - Setup StringGen Spout --------
StringGenSpout spout = new StringGenSpout(100).withFieldName("str");
-
// 2 - Setup HFS Bolt --------
String hdfsUrl = Helper.getStr(topoConf, HDFS_URI);
RecordFormat format = new LineWriter("str");
SyncPolicy syncPolicy = new CountSyncPolicy(hdfsBatch);
- FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(1.0f,
FileSizeRotationPolicy.Units.GB);
+ FileRotationPolicy rotationPolicy = new FileSizeRotationPolicy(1.0f,
+ FileSizeRotationPolicy.Units.GB);
final int spoutNum = Helper.getInt(topoConf, SPOUT_NUM,
DEFAULT_SPOUT_NUM);
final int boltNum = Helper.getInt(topoConf, BOLT_NUM,
DEFAULT_BOLT_NUM);
// Use default, Storm-generated file names
- FileNameFormat fileNameFormat = new
DefaultFileNameFormat().withPath(Helper.getStr(topoConf, HDFS_PATH));
+ FileNameFormat fileNameFormat = new
DefaultFileNameFormat().withPath(Helper.getStr(topoConf,
+ HDFS_PATH));
// Instantiate the HdfsBolt
HdfsBolt bolt = new HdfsBolt()
- .withFsUrl(hdfsUrl)
- .withFileNameFormat(fileNameFormat)
- .withRecordFormat(format)
- .withRotationPolicy(rotationPolicy)
- .withSyncPolicy(syncPolicy);
-
+ .withFsUrl(hdfsUrl)
+ .withFileNameFormat(fileNameFormat)
+ .withRecordFormat(format)
+ .withRotationPolicy(rotationPolicy)
+ .withSyncPolicy(syncPolicy);
// 3 - Setup Topology --------
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout(SPOUT_ID, spout, spoutNum);
builder.setBolt(BOLT_ID, bolt, boltNum)
- .localOrShuffleGrouping(SPOUT_ID);
+ .localOrShuffleGrouping(SPOUT_ID);
return builder.createTopology();
}
-
/**
* Spout generates random strings and HDFS bolt writes them to a text file.
*/
public static void main(String[] args) throws Exception {
String confFile = "conf/HdfsSpoutTopo.yaml";
- int runTime = -1; //Run until Ctrl-C
+ int runTime = -1; // Run until Ctrl-C
if (args.length > 0) {
runTime = Integer.parseInt(args[0]);
}
@@ -124,7 +121,8 @@ public class StrGenSpoutHdfsBoltTopo {
Map<String, Object> topoConf = Utils.findAndReadConfigFile(confFile);
topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
- topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
"org.apache.storm.policy.WaitStrategyPark");
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
+ "org.apache.storm.policy.WaitStrategyPark");
topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
@@ -133,7 +131,6 @@ public class StrGenSpoutHdfsBoltTopo {
Helper.runOnClusterAndPrintMetrics(runTime, TOPOLOGY_NAME, topoConf,
getTopology(topoConf));
}
-
public static class LineWriter implements RecordFormat {
private static final long serialVersionUID = 7524288317405514146L;
private String lineDelimiter = System.lineSeparator();
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ThroughputMeter.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ThroughputMeter.java
index 3d8e73622..8a7b439b7 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/ThroughputMeter.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/ThroughputMeter.java
@@ -32,6 +32,7 @@ public class ThroughputMeter {
/**
* Calculate throughput.
+ *
* @return events/sec
*/
private static double calcThroughput(long count, long startTime, long
endTime) {
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/CountBolt.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/CountBolt.java
index ee1bf7ee1..0a4f0e823 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/CountBolt.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/CountBolt.java
@@ -28,7 +28,6 @@ import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
-
public class CountBolt extends BaseBasicBolt {
public static final String FIELDS_WORD = "word";
public static final String FIELDS_COUNT = "count";
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/DevNullBolt.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/DevNullBolt.java
index cc181c04c..afffa9d01 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/DevNullBolt.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/DevNullBolt.java
@@ -28,7 +28,6 @@ import org.apache.storm.tuple.Tuple;
import org.apache.storm.utils.ObjectReader;
import org.slf4j.LoggerFactory;
-
public class DevNullBolt extends BaseRichBolt {
private static final org.slf4j.Logger LOG =
LoggerFactory.getLogger(DevNullBolt.class);
private OutputCollector collector;
@@ -36,7 +35,8 @@ public class DevNullBolt extends BaseRichBolt {
private int count = 0;
@Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context,
OutputCollector collector) {
+ public void prepare(Map<String, Object> topoConf, TopologyContext context,
+ OutputCollector collector) {
this.collector = collector;
this.sleepNanos =
ObjectReader.getLong(topoConf.get("nullbolt.sleep.micros"), 0L) * 1_000;
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/IdBolt.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/IdBolt.java
index 5387499bb..d0d9de022 100644
--- a/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/IdBolt.java
+++ b/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/IdBolt.java
@@ -31,7 +31,8 @@ public class IdBolt extends BaseRichBolt {
private OutputCollector collector;
@Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context,
OutputCollector collector) {
+ public void prepare(Map<String, Object> topoConf, TopologyContext context,
+ OutputCollector collector) {
this.collector = collector;
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/SplitSentenceBolt.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/SplitSentenceBolt.java
index 85a3f5aa9..64e002b48 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/SplitSentenceBolt.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/bolt/SplitSentenceBolt.java
@@ -27,7 +27,6 @@ import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
-
public class SplitSentenceBolt extends BaseBasicBolt {
public static final String FIELDS = "word";
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/queuetest/Acker.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/queuetest/Acker.java
index adcc3a3d6..a5635bffe 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/queuetest/Acker.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/queuetest/Acker.java
@@ -33,7 +33,6 @@ class Acker extends MyThread {
this.spoutInQ = spoutInQ;
}
-
@Override
public void run() {
long start = System.currentTimeMillis();
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/queuetest/JCQueuePerfTest.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/queuetest/JCQueuePerfTest.java
index 2cbee19e3..883dc87cd 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/queuetest/JCQueuePerfTest.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/queuetest/JCQueuePerfTest.java
@@ -47,8 +47,10 @@ public class JCQueuePerfTest {
private static void ackingProducerSimulation() {
WaitStrategyPark ws = new WaitStrategyPark(100);
StormMetricRegistry registry = new StormMetricRegistry();
- JCQueue spoutQ = new JCQueue("spoutQ", "spoutQ", 1024, 0, 100, ws,
"test", "test", Collections.singletonList(1000), 1000, registry);
- JCQueue ackQ = new JCQueue("ackQ", "ackQ", 1024, 0, 100, ws, "test",
"test", Collections.singletonList(1000), 1000, registry);
+ JCQueue spoutQ = new JCQueue("spoutQ", "spoutQ", 1024, 0, 100, ws,
"test", "test",
+ Collections.singletonList(1000), 1000, registry);
+ JCQueue ackQ = new JCQueue("ackQ", "ackQ", 1024, 0, 100, ws, "test",
"test", Collections
+ .singletonList(1000), 1000, registry);
final AckingProducer ackingProducer = new AckingProducer(spoutQ, ackQ);
final Acker acker = new Acker(ackQ, spoutQ);
@@ -61,7 +63,8 @@ public class JCQueuePerfTest {
StormMetricRegistry registry = new StormMetricRegistry();
JCQueue q1 = new JCQueue("q1", "q1", 1024, 0, prodBatchSz, ws, "test",
"test",
Collections.singletonList(1000), 1000, registry);
- JCQueue q2 = new JCQueue("q2", "q2", 1024, 0, prodBatchSz, ws, "test",
"test", Collections.singletonList(1000), 1000, registry);
+ JCQueue q2 = new JCQueue("q2", "q2", 1024, 0, prodBatchSz, ws, "test",
"test", Collections
+ .singletonList(1000), 1000, registry);
final Producer prod = new Producer(q1);
final Forwarder fwd = new Forwarder(q1, q2);
@@ -70,9 +73,9 @@ public class JCQueuePerfTest {
runAllThds(prod, fwd, cons);
}
-
private static void oneProducer1Consumer(int prodBatchSz) {
- JCQueue q1 = new JCQueue("q1", "q1", 50_000, 0, prodBatchSz, new
WaitStrategyPark(100), "test", "test",
+ JCQueue q1 = new JCQueue("q1", "q1", 50_000, 0, prodBatchSz, new
WaitStrategyPark(100),
+ "test", "test",
Collections.singletonList(1000), 1000, new
StormMetricRegistry());
final Producer prod1 = new Producer(q1);
@@ -82,7 +85,8 @@ public class JCQueuePerfTest {
}
private static void twoProducer1Consumer(int prodBatchSz) {
- JCQueue q1 = new JCQueue("q1", "q1", 50_000, 0, prodBatchSz, new
WaitStrategyPark(100), "test", "test",
+ JCQueue q1 = new JCQueue("q1", "q1", 50_000, 0, prodBatchSz, new
WaitStrategyPark(100),
+ "test", "test",
Collections.singletonList(1000), 1000, new
StormMetricRegistry());
final Producer prod1 = new Producer(q1);
@@ -93,7 +97,8 @@ public class JCQueuePerfTest {
}
private static void threeProducer1Consumer(int prodBatchSz) {
- JCQueue q1 = new JCQueue("q1", "q1", 50_000, 0, prodBatchSz, new
WaitStrategyPark(100), "test", "test",
+ JCQueue q1 = new JCQueue("q1", "q1", 50_000, 0, prodBatchSz, new
WaitStrategyPark(100),
+ "test", "test",
Collections.singletonList(1000), 1000, new
StormMetricRegistry());
final Producer prod1 = new Producer(q1);
@@ -104,12 +109,13 @@ public class JCQueuePerfTest {
runAllThds(prod1, prod2, prod3, cons1);
}
-
private static void oneProducer2Consumers(int prodBatchSz) {
WaitStrategyPark ws = new WaitStrategyPark(100);
StormMetricRegistry registry = new StormMetricRegistry();
- JCQueue q1 = new JCQueue("q1", "q1", 1024, 0, prodBatchSz, ws, "test",
"test", Collections.singletonList(1000), 1000, registry);
- JCQueue q2 = new JCQueue("q2", "q2", 1024, 0, prodBatchSz, ws, "test",
"test", Collections.singletonList(1000), 1000, registry);
+ JCQueue q1 = new JCQueue("q1", "q1", 1024, 0, prodBatchSz, ws, "test",
"test", Collections
+ .singletonList(1000), 1000, registry);
+ JCQueue q2 = new JCQueue("q2", "q2", 1024, 0, prodBatchSz, ws, "test",
"test", Collections
+ .singletonList(1000), 1000, registry);
final Producer2 prod1 = new Producer2(q1, q2);
final Consumer cons1 = new Consumer(q1);
@@ -140,7 +146,8 @@ public class JCQueuePerfTest {
}
for (MyThread thread : threads) {
- System.err.printf("%s : %d, Throughput: %,d \n",
thread.getName(), thread.count, thread.throughput());
+ System.err.printf("%s : %d, Throughput: %,d \n",
thread.getName(),
+ thread.count, thread.throughput());
}
} catch (InterruptedException e) {
return;
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/ConstSpout.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/ConstSpout.java
index 656f992ef..49154bc88 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/ConstSpout.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/ConstSpout.java
@@ -52,7 +52,8 @@ public class ConstSpout extends BaseRichSpout {
}
@Override
- public void open(Map<String, Object> conf, TopologyContext context,
SpoutOutputCollector collector) {
+ public void open(Map<String, Object> conf, TopologyContext context,
+ SpoutOutputCollector collector) {
this.collector = collector;
this.sleep = ObjectReader.getLong(conf.get("spout.sleep"), 0L);
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/FileReadSpout.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/FileReadSpout.java
index a819fa25e..bdfb27fa3 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/FileReadSpout.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/FileReadSpout.java
@@ -44,7 +44,6 @@ public class FileReadSpout extends BaseRichSpout {
private long count = 0;
-
public FileReadSpout(String file) {
this.file = file;
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/StringGenSpout.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/StringGenSpout.java
index 1d0470c3e..5139a33fd 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/StringGenSpout.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/StringGenSpout.java
@@ -30,7 +30,8 @@ import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
/**
- * Spout pre-computes a list with 30k fixed length random strings. Emits
sequentially from this list, over and over again.
+ * Spout pre-computes a list with 30k fixed length random strings. Emits
sequentially from this
+ * list, over and over again.
*/
public class StringGenSpout extends BaseRichSpout {
@@ -67,7 +68,8 @@ public class StringGenSpout extends BaseRichSpout {
}
@Override
- public void open(Map<String, Object> conf, TopologyContext context,
SpoutOutputCollector collector) {
+ public void open(Map<String, Object> conf, TopologyContext context,
+ SpoutOutputCollector collector) {
this.records = genStringList(strLen, strCount);
this.collector = collector;
@@ -83,7 +85,6 @@ public class StringGenSpout extends BaseRichSpout {
}
}
-
@Override
public void ack(Object msgId) {
super.ack(msgId);
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/WordGenSpout.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/WordGenSpout.java
index a6a1fc9b3..c7ea29ac9 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/WordGenSpout.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/spout/WordGenSpout.java
@@ -45,7 +45,6 @@ public class WordGenSpout extends BaseRichSpout {
private ThroughputMeter emitMeter;
private ArrayList<String> words;
-
public WordGenSpout(String file) {
this.file = file;
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/toolstest/JCToolsPerfTest.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/toolstest/JCToolsPerfTest.java
index b5afe28b6..492a94549 100644
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/toolstest/JCToolsPerfTest.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/toolstest/JCToolsPerfTest.java
@@ -73,7 +73,6 @@ public class JCToolsPerfTest {
runAllThds(prod1, prod2, prod3, cons1);
}
-
private static void oneProducer2Consumers() {
MpscArrayQueue<Object> q1 = new MpscArrayQueue<Object>(50_000);
MpscArrayQueue<Object> q2 = new MpscArrayQueue<Object>(50_000);
@@ -107,7 +106,8 @@ public class JCToolsPerfTest {
}
for (MyThd thread : threads) {
- System.err.printf("%s : %d, Throughput: %,d \n",
thread.getName(), thread.count, thread.throughput());
+ System.err.printf("%s : %d, Throughput: %,d \n",
thread.getName(),
+ thread.count, thread.throughput());
}
} catch (InterruptedException e) {
return;
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/BasicMetricsCollector.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/BasicMetricsCollector.java
index 735abe01a..e3b5618f5 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/BasicMetricsCollector.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/BasicMetricsCollector.java
@@ -109,7 +109,7 @@ public class BasicMetricsCollector implements AutoCloseable
{
}
boolean updateStats(PrintWriter writer)
- throws Exception {
+ throws Exception {
if (collectTopologyStats) {
updateTopologyStats();
}
@@ -132,10 +132,12 @@ public class BasicMetricsCollector implements
AutoCloseable {
void updateExecutorStats() {
long timeDiff = this.curSample.getSampleTime() -
this.lastSample.getSampleTime();
- long transferredDiff = this.curSample.getTotalTransferred() -
this.lastSample.getTotalTransferred();
+ long transferredDiff = this.curSample.getTotalTransferred() -
this.lastSample
+ .getTotalTransferred();
long throughput = transferredDiff / (timeDiff / 1000);
- long spoutDiff = this.curSample.getSpoutTransferred() -
this.lastSample.getSpoutTransferred();
+ long spoutDiff = this.curSample.getSpoutTransferred() - this.lastSample
+ .getSpoutTransferred();
long spoutAckedDiff = this.curSample.getTotalAcked() -
this.lastSample.getTotalAcked();
long spoutThroughput = spoutDiff / (timeDiff / 1000);
@@ -152,16 +154,15 @@ public class BasicMetricsCollector implements
AutoCloseable {
metrics.put(SPOUT_THROUGHPUT, Long.toString(spoutThroughput));
}
-
if (collectSpoutLatency) {
double latency = this.curSample.getTotalLatency();
if (latency > this.maxLatency) {
this.maxLatency = latency;
}
metrics.put(SPOUT_AVG_COMPLETE_LATENCY,
- String.format(SPOUT_AVG_LATENCY_FORMAT, latency));
+ String.format(SPOUT_AVG_LATENCY_FORMAT, latency));
metrics.put(SPOUT_MAX_COMPLETE_LATENCY,
- String.format(SPOUT_MAX_LATENCY_FORMAT,
this.maxLatency));
+ String.format(SPOUT_MAX_LATENCY_FORMAT, this.maxLatency));
}
}
@@ -192,11 +193,13 @@ public class BasicMetricsCollector implements
AutoCloseable {
}
writer.println(
-
"\n------------------------------------------------------------------------------------------------------------------");
+
"\n-------------------------------------------------------------------------------"
+ + "-----------------------------------");
String str = Utils.join(header, ",");
writer.println(str);
writer
-
.println("------------------------------------------------------------------------------------------------------------------");
+
.println("------------------------------------------------------------------------"
+ + "------------------------------------------");
writer.flush();
}
@@ -212,28 +215,28 @@ public class BasicMetricsCollector implements
AutoCloseable {
boolean collectTopologyStats(Set<MetricsItem> items) {
return items.contains(MetricsItem.ALL)
- || items.contains(MetricsItem.TOPOLOGY_STATS);
+ || items.contains(MetricsItem.TOPOLOGY_STATS);
}
boolean collectExecutorStats(Set<MetricsItem> items) {
return items.contains(MetricsItem.ALL)
- || items.contains(MetricsItem.XSFER_RATE)
- || items.contains(MetricsItem.SPOUT_LATENCY);
+ || items.contains(MetricsItem.XSFER_RATE)
+ || items.contains(MetricsItem.SPOUT_LATENCY);
}
boolean collectThroughput(Set<MetricsItem> items) {
return items.contains(MetricsItem.ALL)
- || items.contains(MetricsItem.XSFER_RATE);
+ || items.contains(MetricsItem.XSFER_RATE);
}
boolean collectSpoutThroughput(Set<MetricsItem> items) {
return items.contains(MetricsItem.ALL)
- || items.contains(MetricsItem.SPOUT_THROUGHPUT);
+ || items.contains(MetricsItem.SPOUT_THROUGHPUT);
}
boolean collectSpoutLatency(Set<MetricsItem> items) {
return items.contains(MetricsItem.ALL)
- || items.contains(MetricsItem.SPOUT_LATENCY);
+ || items.contains(MetricsItem.SPOUT_LATENCY);
}
public enum MetricsItem {
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/Helper.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/Helper.java
index a573308b6..b685279c0 100755
--- a/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/Helper.java
+++ b/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/Helper.java
@@ -27,7 +27,6 @@ import org.apache.storm.utils.NimbusClient;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Utils;
-
public class Helper {
public static void kill(Nimbus.Iface client, String topoName) throws
Exception {
@@ -44,10 +43,12 @@ public class Helper {
return (String) map.get(key);
}
- public static void collectMetricsAndKill(String topologyName, Integer
pollInterval, int duration) throws Exception {
+ public static void collectMetricsAndKill(String topologyName, Integer
pollInterval,
+ int duration) throws Exception {
Map<String, Object> clusterConf = Utils.readStormConfig();
Nimbus.Iface client =
NimbusClient.Builder.withConf(clusterConf).build().getClient();
- try (BasicMetricsCollector metricsCollector = new
BasicMetricsCollector(topologyName, clusterConf)) {
+ try (BasicMetricsCollector metricsCollector = new
BasicMetricsCollector(topologyName,
+ clusterConf)) {
if (duration > 0) {
int times = duration / pollInterval;
@@ -57,7 +58,7 @@ public class Helper {
metricsCollector.collect(client);
}
} else {
- while (true) { //until Ctrl-C
+ while (true) { // until Ctrl-C
metricsCollector.collect(client);
Thread.sleep(pollInterval * 1000);
}
@@ -87,8 +88,9 @@ public class Helper {
});
}
- public static void runOnClusterAndPrintMetrics(int durationSec, String
topoName, Map<String, Object> topoConf, StormTopology topology)
- throws Exception {
+ public static void runOnClusterAndPrintMetrics(int durationSec, String
topoName, Map<String,
+ Object> topoConf,
StormTopology topology)
+ throws Exception {
// submit topology
StormSubmitter.submitTopologyWithProgressBar(topoName, topoConf,
topology);
setupShutdownHook(topoName); // handle Ctrl-C
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/IdentityBolt.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/IdentityBolt.java
index df27263b1..0b1e6f4e1 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/IdentityBolt.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/IdentityBolt.java
@@ -29,7 +29,8 @@ public class IdentityBolt extends BaseRichBolt {
private OutputCollector collector;
@Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context,
OutputCollector collector) {
+ public void prepare(Map<String, Object> topoConf, TopologyContext context,
+ OutputCollector collector) {
this.collector = collector;
}
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/MetricsSample.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/MetricsSample.java
index c411844a3..a6a4ff9cd 100755
---
a/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/MetricsSample.java
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/utils/MetricsSample.java
@@ -20,7 +20,6 @@ package org.apache.storm.perf.utils;
import java.util.List;
import java.util.Map;
-import org.apache.storm.generated.ClusterSummary;
import org.apache.storm.generated.ExecutorSpecificStats;
import org.apache.storm.generated.ExecutorStats;
import org.apache.storm.generated.ExecutorSummary;
@@ -131,7 +130,7 @@ public class MetricsSample {
}
}
- Double total = 0d;
+ Double total = 0D;
Map<String, Double> vals =
spoutStats.get_complete_ms_avg().get(":all-time");
if (vals != null) {
for (String key : vals.keySet()) {
@@ -144,7 +143,6 @@ public class MetricsSample {
spoutExecCount++;
}
-
} // end executor summary
MetricsSample ret = new MetricsSample();
