This is an automated email from the ASF dual-hosted git repository.
srdo 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 e3825f3 STORM-3448: core: fix all checkstyle warnings
new 3ecaf62 Merge pull request #3063 from krichter722/checkstyle-core
e3825f3 is described below
commit e3825f338347a027a0cab190c79bbb3c4f3a0892
Author: Karl-Philipp Richter <krich...@posteo.de>
AuthorDate: Mon Jul 1 20:57:32 2019 +0200
STORM-3448: core: fix all checkstyle warnings
---
storm-core/pom.xml | 2 +-
.../org/apache/storm/command/AdminCommands.java | 19 +++---
.../src/jvm/org/apache/storm/command/CLI.java | 2 +
.../org/apache/storm/command/ListTopologies.java | 9 ++-
.../src/jvm/org/apache/storm/command/Monitor.java | 19 +++---
.../jvm/org/apache/storm/command/Rebalance.java | 9 ++-
.../org/apache/storm/command/ShellSubmission.java | 6 +-
.../org/apache/storm/planner/CompoundSpout.java | 6 +-
.../jvm/org/apache/storm/planner/CompoundTask.java | 6 +-
.../src/jvm/org/apache/storm/utils/Monitor.java | 76 ++++++++++++----------
.../org/apache/storm/utils/TopologySpoutLag.java | 18 ++---
11 files changed, 88 insertions(+), 84 deletions(-)
diff --git a/storm-core/pom.xml b/storm-core/pom.xml
index 2bebab4..71d220f 100644
--- a/storm-core/pom.xml
+++ b/storm-core/pom.xml
@@ -303,7 +303,7 @@
<artifactId>maven-checkstyle-plugin</artifactId>
<!--Note - the version would be inherited-->
<configuration>
- <maxAllowedViolations>73</maxAllowedViolations>
+ <maxAllowedViolations>0</maxAllowedViolations>
</configuration>
</plugin>
<plugin>
diff --git a/storm-core/src/jvm/org/apache/storm/command/AdminCommands.java
b/storm-core/src/jvm/org/apache/storm/command/AdminCommands.java
index 564a01c..8076b74 100644
--- a/storm-core/src/jvm/org/apache/storm/command/AdminCommands.java
+++ b/storm-core/src/jvm/org/apache/storm/command/AdminCommands.java
@@ -72,7 +72,8 @@ public class AdminCommands {
@Override
public void run(String[] args, Map<String, Object> conf, String
command) throws Exception {
try (BlobStore nimbusBlobStore =
ServerUtils.getNimbusBlobStore(conf, NimbusInfo.fromConf(conf), null)) {
- IStormClusterState stormClusterState =
ClusterUtils.mkStormClusterState(conf, new
ClusterStateContext(DaemonType.NIMBUS, conf));
+ IStormClusterState stormClusterState =
ClusterUtils.mkStormClusterState(conf,
+ new ClusterStateContext(DaemonType.NIMBUS, conf));
Set<String> blobStoreTopologyIds =
nimbusBlobStore.filterAndListKeys(key -> ConfigUtils.getIdFromBlobKey(key));
Set<String> activeTopologyIds = new
HashSet<>(stormClusterState.activeStorms());
@@ -126,14 +127,6 @@ public class AdminCommands {
return builder.toString();
}
- private static void println(StringBuilder out, int depth, Object value) {
- for (int i = 0; i < depth; i++) {
- out.append("\t");
- }
- out.append(value);
- out.append("\n");
- }
-
private static void prettyPrint(TBase value, int depth, StringBuilder out)
{
if (value == null) {
println(out, depth,"null");
@@ -144,6 +137,14 @@ public class AdminCommands {
println(out, depth, "}");
}
+ private static void println(StringBuilder out, int depth, Object value) {
+ for (int i = 0; i < depth; i++) {
+ out.append("\t");
+ }
+ out.append(value);
+ out.append("\n");
+ }
+
private static void prettyPrintFields(TBase value, int depth,
StringBuilder out) {
for (Map.Entry<? extends TFieldIdEnum, FieldMetaData> entry :
FieldMetaData.getStructMetaDataMap(value.getClass()).entrySet()) {
TFieldIdEnum key = entry.getKey();
diff --git a/storm-core/src/jvm/org/apache/storm/command/CLI.java
b/storm-core/src/jvm/org/apache/storm/command/CLI.java
index 4510d29..cc5c5dc 100644
--- a/storm-core/src/jvm/org/apache/storm/command/CLI.java
+++ b/storm-core/src/jvm/org/apache/storm/command/CLI.java
@@ -23,6 +23,7 @@ import org.apache.commons.cli.Options;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@SuppressWarnings("checkstyle:AbbreviationAsWordInName")
public class CLI {
/**
* Parse function to return an Integer.
@@ -250,6 +251,7 @@ public class CLI {
}
}
+ @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
public static class CLIBuilder {
private final ArrayList<Opt> opts = new ArrayList<>();
private final ArrayList<Arg> args = new ArrayList<>();
diff --git a/storm-core/src/jvm/org/apache/storm/command/ListTopologies.java
b/storm-core/src/jvm/org/apache/storm/command/ListTopologies.java
index 13d9691..09068d3 100644
--- a/storm-core/src/jvm/org/apache/storm/command/ListTopologies.java
+++ b/storm-core/src/jvm/org/apache/storm/command/ListTopologies.java
@@ -31,7 +31,14 @@ public class ListTopologies {
if (topologies == null || topologies.isEmpty()) {
System.out.println("No topologies running.");
} else {
- System.out.printf(MSG_FORMAT, "Topology_name", "Status",
"Num_tasks", "Num_workers", "Uptime_secs", "Topology_Id", "Owner");
+ System.out.printf(MSG_FORMAT,
+ "Topology_name",
+ "Status",
+ "Num_tasks",
+ "Num_workers",
+ "Uptime_secs",
+ "Topology_Id",
+ "Owner");
System.out.println("----------------------------------------------------------------------------------------");
for (TopologySummary topology : topologies) {
System.out.printf(MSG_FORMAT, topology.get_name(),
topology.get_status(),
diff --git a/storm-core/src/jvm/org/apache/storm/command/Monitor.java
b/storm-core/src/jvm/org/apache/storm/command/Monitor.java
index 7ade55e..6fa69de 100644
--- a/storm-core/src/jvm/org/apache/storm/command/Monitor.java
+++ b/storm-core/src/jvm/org/apache/storm/command/Monitor.java
@@ -27,25 +27,24 @@ public class Monitor {
.parse(args);
final org.apache.storm.utils.Monitor monitor = new
org.apache.storm.utils.Monitor();
Integer interval = (Integer) cl.get("i");
- String component = (String) cl.get("m");
- String stream = (String) cl.get("s");
- String watch = (String) cl.get("w");
- String topologyName = (String) cl.get("topologyName");
-
if (null != interval) {
- monitor.set_interval(interval);
+ monitor.setInterval(interval);
}
+ String component = (String) cl.get("m");
if (null != component) {
- monitor.set_component(component);
+ monitor.setComponent(component);
}
+ String stream = (String) cl.get("s");
if (null != stream) {
- monitor.set_stream(stream);
+ monitor.setStream(stream);
}
+ String watch = (String) cl.get("w");
if (null != watch) {
- monitor.set_watch(watch);
+ monitor.setWatch(watch);
}
+ String topologyName = (String) cl.get("topologyName");
if (null != topologyName) {
- monitor.set_topology(topologyName);
+ monitor.setTopology(topologyName);
}
NimbusClient.withConfiguredClient(new NimbusClient.WithNimbus() {
diff --git a/storm-core/src/jvm/org/apache/storm/command/Rebalance.java
b/storm-core/src/jvm/org/apache/storm/command/Rebalance.java
index 2ca7630..92dcc7b 100644
--- a/storm-core/src/jvm/org/apache/storm/command/Rebalance.java
+++ b/storm-core/src/jvm/org/apache/storm/command/Rebalance.java
@@ -13,6 +13,7 @@
package org.apache.storm.command;
import static java.lang.String.format;
+
import java.util.HashMap;
import java.util.Map;
import org.apache.storm.generated.Nimbus;
@@ -39,20 +40,18 @@ public class Rebalance {
final String name = (String) cl.get("topologyName");
final RebalanceOptions rebalanceOptions = new RebalanceOptions();
Integer wait = (Integer) cl.get("w");
- Integer numWorkers = (Integer) cl.get("n");
- Map<String, Integer> numExecutors = (Map<String, Integer>) cl.get("e");
- Map<String, Map<String, Double>> resourceOverrides = (Map<String,
Map<String, Double>>) cl.get("r");
-
if (null != wait) {
rebalanceOptions.set_wait_secs(wait);
}
+ Integer numWorkers = (Integer) cl.get("n");
if (null != numWorkers) {
rebalanceOptions.set_num_workers(numWorkers);
}
+ Map<String, Integer> numExecutors = (Map<String, Integer>) cl.get("e");
if (null != numExecutors) {
rebalanceOptions.set_num_executors(numExecutors);
}
-
+ Map<String, Map<String, Double>> resourceOverrides = (Map<String,
Map<String, Double>>) cl.get("r");
if (null != resourceOverrides) {
rebalanceOptions.set_topology_resources_overrides(resourceOverrides);
}
diff --git a/storm-core/src/jvm/org/apache/storm/command/ShellSubmission.java
b/storm-core/src/jvm/org/apache/storm/command/ShellSubmission.java
index 895a266..0ef320c 100644
--- a/storm-core/src/jvm/org/apache/storm/command/ShellSubmission.java
+++ b/storm-core/src/jvm/org/apache/storm/command/ShellSubmission.java
@@ -12,6 +12,9 @@
package org.apache.storm.command;
+import java.util.Arrays;
+import java.util.Map;
+
import org.apache.commons.lang.ArrayUtils;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.NimbusSummary;
@@ -21,9 +24,6 @@ import org.apache.storm.utils.ServerUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Arrays;
-import java.util.Map;
-
public class ShellSubmission {
private static final Logger LOG =
LoggerFactory.getLogger(ShellSubmission.class);
diff --git a/storm-core/src/jvm/org/apache/storm/planner/CompoundSpout.java
b/storm-core/src/jvm/org/apache/storm/planner/CompoundSpout.java
index 2a46f27..6db0de7 100644
--- a/storm-core/src/jvm/org/apache/storm/planner/CompoundSpout.java
+++ b/storm-core/src/jvm/org/apache/storm/planner/CompoundSpout.java
@@ -12,9 +12,5 @@
package org.apache.storm.planner;
-
-public class CompoundSpout
- //implements ISpout
-{
-
+public class CompoundSpout {
}
diff --git a/storm-core/src/jvm/org/apache/storm/planner/CompoundTask.java
b/storm-core/src/jvm/org/apache/storm/planner/CompoundTask.java
index c9fe535..c5b4b0b 100644
--- a/storm-core/src/jvm/org/apache/storm/planner/CompoundTask.java
+++ b/storm-core/src/jvm/org/apache/storm/planner/CompoundTask.java
@@ -12,9 +12,5 @@
package org.apache.storm.planner;
-
-public class CompoundTask
- // implements IBolt
-{
-
+public class CompoundTask {
}
diff --git a/storm-core/src/jvm/org/apache/storm/utils/Monitor.java
b/storm-core/src/jvm/org/apache/storm/utils/Monitor.java
index 9ff2d36..f206802 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/Monitor.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/Monitor.java
@@ -27,11 +27,11 @@ public class Monitor {
private static final String WATCH_TRANSFERRED = "transferred";
private static final String WATCH_EMITTED = "emitted";
- private int _interval = 4;
- private String _topology;
- private String _component;
- private String _stream;
- private String _watch;
+ private int interval = 4;
+ private String topology;
+ private String component;
+ private String stream;
+ private String watch;
private HashSet<String> getComponents(Nimbus.Iface client, String
topology) throws Exception {
HashSet<String> components = new HashSet<>();
@@ -58,17 +58,17 @@ public class Monitor {
}
public void metrics(Nimbus.Iface client) throws Exception {
- if (_interval <= 0) {
+ if (interval <= 0) {
throw new IllegalArgumentException("poll interval must be
positive");
}
- if (_topology == null || _topology.isEmpty()) {
+ if (topology == null || topology.isEmpty()) {
throw new IllegalArgumentException("topology name must be
something");
}
- if (_component == null || _component.isEmpty()) {
- HashSet<String> components = getComponents(client, _topology);
- System.out.println("Available components for " + _topology + " :");
+ if (component == null || component.isEmpty()) {
+ HashSet<String> components = getComponents(client, topology);
+ System.out.println("Available components for " + topology + " :");
System.out.println("------------------");
for (String comp : components) {
System.out.println(comp);
@@ -78,16 +78,16 @@ public class Monitor {
return;
}
- if (_stream == null || _stream.isEmpty()) {
+ if (stream == null || stream.isEmpty()) {
throw new IllegalArgumentException("stream name must be
something");
}
- if (!WATCH_TRANSFERRED.equals(_watch) &&
!WATCH_EMITTED.equals(_watch)) {
+ if (!WATCH_TRANSFERRED.equals(watch) && !WATCH_EMITTED.equals(watch)) {
throw new IllegalArgumentException("watch item must either be
transferred or emitted");
}
-
System.out.println("topology\tcomponent\tparallelism\tstream\ttime-diff ms\t" +
_watch + "\tthroughput (Kt/s)");
+
System.out.println("topology\tcomponent\tparallelism\tstream\ttime-diff ms\t" +
watch + "\tthroughput (Kt/s)");
- long pollMs = _interval * 1000;
+ long pollMs = interval * 1000;
long now = System.currentTimeMillis();
MetricsState state = new MetricsState(now, 0);
Poller poller = new Poller(now, pollMs);
@@ -111,29 +111,29 @@ public class Monitor {
ClusterSummary clusterSummary = client.getClusterInfo();
TopologySummary topologySummary = null;
for (TopologySummary ts : clusterSummary.get_topologies()) {
- if (_topology.equals(ts.get_name())) {
+ if (topology.equals(ts.get_name())) {
topologySummary = ts;
break;
}
}
if (topologySummary == null) {
- throw new IllegalArgumentException("topology: " + _topology + "
not found");
+ throw new IllegalArgumentException("topology: " + topology + " not
found");
} else {
String id = topologySummary.get_id();
GetInfoOptions getInfoOpts = new GetInfoOptions();
getInfoOpts.set_num_err_choice(NumErrorsChoice.NONE);
TopologyInfo info = client.getTopologyInfoWithOpts(id,
getInfoOpts);
for (ExecutorSummary es : info.get_executors()) {
- if (_component.equals(es.get_component_id())) {
+ if (component.equals(es.get_component_id())) {
componentParallelism++;
ExecutorStats stats = es.get_stats();
if (stats != null) {
Map<String, Map<String, Long>> statted =
- WATCH_EMITTED.equals(_watch) ? stats.get_emitted()
: stats.get_transferred();
+ WATCH_EMITTED.equals(watch) ? stats.get_emitted()
: stats.get_transferred();
if (statted != null) {
Map<String, Long> e2 = statted.get(":all-time");
if (e2 != null) {
- Long stream = e2.get(_stream);
+ Long stream = e2.get(this.stream);
if (stream != null) {
streamFound = true;
totalStatted += stream;
@@ -146,47 +146,51 @@ public class Monitor {
}
if (componentParallelism <= 0) {
- HashSet<String> components = getComponents(client, _topology);
- System.out.println("Available components for " + _topology + " :");
+ HashSet<String> components = getComponents(client, topology);
+ System.out.println("Available components for " + topology + " :");
System.out.println("------------------");
for (String comp : components) {
System.out.println(comp);
}
System.out.println("------------------");
- throw new IllegalArgumentException("component: " + _component + "
not found");
+ throw new IllegalArgumentException("component: " + component + "
not found");
}
if (!streamFound) {
- throw new IllegalArgumentException("stream: " + _stream + " not
found");
+ throw new IllegalArgumentException("stream: " + stream + " not
found");
}
long timeDelta = now - state.getLastTime();
long stattedDelta = totalStatted - state.getLastStatted();
state.setLastTime(now);
state.setLastStatted(totalStatted);
double throughput = (stattedDelta == 0 || timeDelta == 0) ? 0.0 :
((double) stattedDelta / (double) timeDelta);
- System.out.println(
- _topology + "\t" + _component + "\t" + componentParallelism + "\t"
+ _stream + "\t" + timeDelta + "\t" + stattedDelta + "\t" +
- throughput);
+ System.out.println(topology + "\t"
+ + component + "\t"
+ + componentParallelism + "\t"
+ + stream + "\t"
+ + timeDelta + "\t"
+ + stattedDelta + "\t"
+ + throughput);
}
- public void set_interval(int _interval) {
- this._interval = _interval;
+ public void setInterval(int interval) {
+ this.interval = interval;
}
- public void set_topology(String _topology) {
- this._topology = _topology;
+ public void setTopology(String topology) {
+ this.topology = topology;
}
- public void set_component(String _component) {
- this._component = _component;
+ public void setComponent(String component) {
+ this.component = component;
}
- public void set_stream(String _stream) {
- this._stream = _stream;
+ public void setStream(String stream) {
+ this.stream = stream;
}
- public void set_watch(String _watch) {
- this._watch = _watch;
+ public void setWatch(String watch) {
+ this.watch = watch;
}
private static class MetricsState {
diff --git a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
index ceab7a5..ec90d15 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
@@ -16,6 +16,8 @@
package org.apache.storm.utils;
+import com.google.common.base.Strings;
+
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
@@ -36,8 +38,6 @@ import org.json.simple.parser.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Strings;
-
public class TopologySpoutLag {
// FIXME: This class can be moved to webapp once UI porting is done.
@@ -52,7 +52,7 @@ public class TopologySpoutLag {
private static final String SECURITY_PROTOCOL_CONFIG = CONFIG_KEY_PREFIX +
"security.protocol";
private static final Set<String> ALL_CONFIGS = new
HashSet<>(Arrays.asList(TOPICS_CONFIG, GROUPID_CONFIG,
BOOTSTRAP_CONFIG, SECURITY_PROTOCOL_CONFIG));
- private final static Logger logger =
LoggerFactory.getLogger(TopologySpoutLag.class);
+ private static final Logger LOGGER =
LoggerFactory.getLogger(TopologySpoutLag.class);
public static Map<String, Map<String, Object>> lag(StormTopology
stormTopology, Map<String, Object> topologyConf) {
Map<String, Map<String, Object>> result = new HashMap<>();
@@ -62,15 +62,15 @@ public class TopologySpoutLag {
SpoutSpec spoutSpec = spout.getValue();
addLagResultForKafkaSpout(result, spout.getKey(), spoutSpec);
} catch (Exception e) {
- logger.warn("Exception thrown while getting lag for spout id:
" + spout.getKey());
- logger.warn("Exception message:" + e.getMessage(), e);
+ LOGGER.warn("Exception thrown while getting lag for spout id:
" + spout.getKey());
+ LOGGER.warn("Exception message:" + e.getMessage(), e);
}
}
return result;
}
private static List<String>
getCommandLineOptionsForNewKafkaSpout(Map<String, Object> jsonConf) {
- logger.debug("json configuration: {}", jsonConf);
+ LOGGER.debug("json configuration: {}", jsonConf);
List<String> commands = new ArrayList<>();
commands.add("-t");
@@ -101,7 +101,7 @@ public class TopologySpoutLag {
file.deleteOnExit();
Properties properties = new Properties();
properties.putAll(extraProperties);
- try(FileOutputStream fos = new FileOutputStream(file)) {
+ try (FileOutputStream fos = new FileOutputStream(file)) {
properties.store(fos, "Kafka consumer extra properties");
}
} catch (IOException ex) {
@@ -163,7 +163,7 @@ public class TopologySpoutLag {
commands.add("-c");
commands.add(extraPropertiesFile.getAbsolutePath());
}
- logger.debug("Command to run: {}", commands);
+ LOGGER.debug("Command to run: {}", commands);
// if commands contains one or more null value, spout is compiled
with lower version of storm-kafka-client
if (!commands.contains(null)) {
@@ -173,7 +173,7 @@ public class TopologySpoutLag {
try {
result = (Map<String, Object>)
JSONValue.parseWithException(resultFromMonitor);
} catch (ParseException e) {
- logger.debug("JSON parsing failed, assuming message as
error message: {}", resultFromMonitor);
+ LOGGER.debug("JSON parsing failed, assuming message as
error message: {}", resultFromMonitor);
// json parsing fail -> error received
errorMsg = resultFromMonitor;
}
