You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2018/06/01 15:17:31 UTC
[2/3] storm git commit: review rework
review rework
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/787a4340
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/787a4340
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/787a4340
Branch: refs/heads/master
Commit: 787a4340a8d2d3291e5e4ecef47c8b85ff9a3b74
Parents: 99f4757
Author: Aaron Gresch <ag...@yahoo-inc.com>
Authored: Wed May 30 10:40:23 2018 -0500
Committer: Aaron Gresch <ag...@yahoo-inc.com>
Committed: Wed May 30 10:40:23 2018 -0500
----------------------------------------------------------------------
conf/defaults.yaml | 2 ++
.../test/clj/org/apache/storm/nimbus_test.clj | 2 +-
.../java/org/apache/storm/DaemonConfig.java | 2 +-
.../org/apache/storm/daemon/nimbus/Nimbus.java | 24 +++++++++++---------
.../apache/storm/daemon/nimbus/NimbusTest.java | 5 ++--
5 files changed, 20 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/787a4340/conf/defaults.yaml
----------------------------------------------------------------------
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index 0e67957..a304aa5 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -50,6 +50,7 @@ storm.nimbus.retry.interval.millis: 2000
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.cluster.state.store: "org.apache.storm.cluster.ZKStateStorageFactory"
storm.meta.serialization.delegate: "org.apache.storm.serialization.GzipThriftSerializationDelegate"
@@ -83,6 +84,7 @@ nimbus.local.assignments.backend.class: "org.apache.storm.assignments.InMemoryAs
nimbus.assignments.service.threads: 10
nimbus.assignments.service.thread.queue.size: 100
nimbus.worker.heartbeats.recovery.strategy.class: "org.apache.storm.nimbus.TimeOutWorkerHeartbeatsRecoveryStrategy"
+nimbus.topology.blobstore.deletion.delay.ms: 300000
### ui.* configs are for the master
ui.host: 0.0.0.0
http://git-wip-us.apache.org/repos/asf/storm/blob/787a4340/storm-core/test/clj/org/apache/storm/nimbus_test.clj
----------------------------------------------------------------------
diff --git a/storm-core/test/clj/org/apache/storm/nimbus_test.clj b/storm-core/test/clj/org/apache/storm/nimbus_test.clj
index 93246d3..77cbc98 100644
--- a/storm-core/test/clj/org/apache/storm/nimbus_test.clj
+++ b/storm-core/test/clj/org/apache/storm/nimbus_test.clj
@@ -1885,7 +1885,7 @@
hb-cache (into {}(map vector inactive-topos '(nil nil)))
mock-state (mock-cluster-state)
mock-blob-store (Mockito/mock BlobStore)
- conf {NIMBUS-MONITOR-FREQ-SECS 10 NIMBUS-TOPOLOGY-BLOBSTORE-DELETION-DELAY-MSEC 0}]
+ conf {NIMBUS-MONITOR-FREQ-SECS 10 NIMBUS-TOPOLOGY-BLOBSTORE-DELETION-DELAY-MS 0}]
(with-open [_ (MockedZookeeper. (proxy [Zookeeper] []
(zkLeaderElectorImpl [conf zk blob-store tc cluster-state acls] (MockLeaderElector. ))))]
(let [nimbus (Mockito/spy (Nimbus. conf nil mock-state nil mock-blob-store nil nil))]
http://git-wip-us.apache.org/repos/asf/storm/blob/787a4340/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
index 3f04d83..f174773 100644
--- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
+++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
@@ -283,7 +283,7 @@ public class DaemonConfig implements Validated {
* This controls the number of milliseconds nimbus will wait before deleting a topology blobstore once detected it is able to delete.
*/
@isInteger
- public static final String NIMBUS_TOPOLOGY_BLOBSTORE_DELETION_DELAY_MSEC = "nimbus.topology.blobstore.deletion.delay.msec";
+ public static final String NIMBUS_TOPOLOGY_BLOBSTORE_DELETION_DELAY_MS = "nimbus.topology.blobstore.deletion.delay.ms";
/**
* Storm UI binds to this host/interface.
http://git-wip-us.apache.org/repos/asf/storm/blob/787a4340/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index 9e172d0..4bd2f4a 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -372,7 +372,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
private static final List<String> EMPTY_STRING_LIST = Collections.unmodifiableList(Collections.emptyList());
private static final Set<String> EMPTY_STRING_SET = Collections.unmodifiableSet(Collections.emptySet());
private static final Pattern TOPOLOGY_NAME_REGEX = Pattern.compile("^[^/.:\\\\]+$");
- private static final RotatingMap<String, Long> topologyCleanupDetected = new RotatingMap<String, Long>(2);
+ private static final RotatingMap<String, Long> topologyCleanupDetected = new RotatingMap<>(2);
private static long topologyCleanupRotationTime = 0L;
// END TOPOLOGY STATE TRANSITIONS
@@ -848,20 +848,22 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
}
/**
- * Finds blobstore entries with no matching topology. Waits NIMBUS_TOPOLOGY_BLOBSTORE_DELETION_DELAY_MSEC
- * before reporting the topologies found. The delay is to prevent a race condition between when a blobstore
- * is created and when the topology is submitted. It is possible the Nimbus cleanup timer task will find
- * entries to delete between these two events.
+ * Finds blobstore entries with no matching topology. Blobstore entries first detected less than
+ * NIMBUS_TOPOLOGY_BLOBSTORE_DELETION_DELAY_MS ago are ignored. The delay is to prevent a race condition
+ * between when a blobstore is created and when the topology is submitted. It is possible the Nimbus cleanup
+ * timer task will find entries to delete between these two events.
*
- * @param store blobstore to search
- * @param conf the nimbus conf
- * @return a set of
+ * Tracked blobstore entries are rotated out of the stored map periodically.
+ *
+ * @param store blobstore to search
+ * @param conf the nimbus conf
+ * @return the set of blobstores with no matching topology
*/
- static Set<String> getIdleTopologyIds(BlobStore store, Map<String, Object> conf) {
+ static Set<String> getExpiredTopologyIds(BlobStore store, Map<String, Object> conf) {
Set<String> toposToClean = store.storedTopoIds();
Set<String> idleTopologies = new HashSet<>();
long topologyDeletionDelay = ObjectReader.getInt(
- conf.get(DaemonConfig.NIMBUS_TOPOLOGY_BLOBSTORE_DELETION_DELAY_MSEC), 5 * 60 * 1000);
+ conf.get(DaemonConfig.NIMBUS_TOPOLOGY_BLOBSTORE_DELETION_DELAY_MS), 5 * 60 * 1000);
for (String topologyId : toposToClean) {
if (Time.currentTimeMillis() - getTopologyCleanupDetectedTime(topologyId) >= topologyDeletionDelay) {
idleTopologies.add(topologyId);
@@ -878,7 +880,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
Set<String> ret = new HashSet<>();
ret.addAll(Utils.OR(state.heartbeatStorms(), EMPTY_STRING_LIST));
ret.addAll(Utils.OR(state.errorTopologies(), EMPTY_STRING_LIST));
- ret.addAll(Utils.OR(getIdleTopologyIds(store, conf), EMPTY_STRING_SET));
+ ret.addAll(Utils.OR(getExpiredTopologyIds(store, conf), EMPTY_STRING_SET));
ret.addAll(Utils.OR(state.backpressureTopologies(), EMPTY_STRING_LIST));
ret.addAll(Utils.OR(state.idsOfTopologiesWithPrivateWorkerKeys(), EMPTY_STRING_SET));
ret.removeAll(Utils.OR(state.activeStorms(), EMPTY_STRING_LIST));
http://git-wip-us.apache.org/repos/asf/storm/blob/787a4340/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
index bad75f0..8018a81 100644
--- a/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
+++ b/storm-server/src/test/java/org/apache/storm/daemon/nimbus/NimbusTest.java
@@ -72,14 +72,15 @@ public class NimbusTest {
idleTopologies.add("topology1");
Mockito.when(store.storedTopoIds()).thenReturn(idleTopologies);
Map<String, Object> conf = new HashMap<>();
+ conf.put(DaemonConfig.NIMBUS_TOPOLOGY_BLOBSTORE_DELETION_DELAY_MS, 300000);
try (Time.SimulatedTime t = new Time.SimulatedTime(null)) {
- Set<String> toDelete = Nimbus.getIdleTopologyIds(store, conf);
+ Set<String> toDelete = Nimbus.getExpiredTopologyIds(store, conf);
Assert.assertTrue(toDelete.isEmpty());
Time.advanceTime(10 * 60 * 1000L);
- toDelete = Nimbus.getIdleTopologyIds(store, conf);
+ toDelete = Nimbus.getExpiredTopologyIds(store, conf);
Assert.assertTrue(toDelete.contains("topology1"));
Assert.assertEquals(1, toDelete.size());