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:30 UTC
[1/3] storm git commit: STORM-3053 prevent race deleting blobstores
before topologies can be submitted
Repository: storm
Updated Branches:
refs/heads/master 0bc4e26d1 -> b0bdbf082
STORM-3053 prevent race deleting blobstores before topologies can be submitted
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/99f47570
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/99f47570
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/99f47570
Branch: refs/heads/master
Commit: 99f47570750266c12d31eaa11ef912e92575b095
Parents: 66559c6
Author: Aaron Gresch <ag...@yahoo-inc.com>
Authored: Mon May 21 14:52:58 2018 -0500
Committer: Aaron Gresch <ag...@yahoo-inc.com>
Committed: Mon May 21 15:05:41 2018 -0500
----------------------------------------------------------------------
.../test/clj/org/apache/storm/nimbus_test.clj | 10 ++--
.../java/org/apache/storm/DaemonConfig.java | 6 +++
.../org/apache/storm/daemon/nimbus/Nimbus.java | 55 ++++++++++++++++++--
.../apache/storm/daemon/nimbus/NimbusTest.java | 30 +++++++++++
4 files changed, 92 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/99f47570/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 a974231..93246d3 100644
--- a/storm-core/test/clj/org/apache/storm/nimbus_test.clj
+++ b/storm-core/test/clj/org/apache/storm/nimbus_test.clj
@@ -1856,7 +1856,7 @@
(let [mock-state (mock-cluster-state (list "topo1") (list "topo1" "topo2" "topo3"))
store (Mockito/mock BlobStore)]
(.thenReturn (Mockito/when (.storedTopoIds store)) #{})
- (is (= (Nimbus/topoIdsToClean mock-state store) #{"topo2" "topo3"}))))
+ (is (= (Nimbus/topoIdsToClean mock-state store (new java.util.HashMap)) #{"topo2" "topo3"}))))
(deftest cleanup-storm-ids-performs-union-of-storm-ids-with-active-znodes
(let [active-topos (list "hb1" "e2" "bp3")
@@ -1866,7 +1866,7 @@
mock-state (mock-cluster-state active-topos hb-topos error-topos bp-topos)
store (Mockito/mock BlobStore)]
(.thenReturn (Mockito/when (.storedTopoIds store)) #{})
- (is (= (Nimbus/topoIdsToClean mock-state store)
+ (is (= (Nimbus/topoIdsToClean mock-state store (new java.util.HashMap))
#{"hb2" "hb3" "e1" "e3" "bp1" "bp2"}))))
(deftest cleanup-storm-ids-returns-empty-set-when-all-topos-are-active
@@ -1877,7 +1877,7 @@
mock-state (mock-cluster-state active-topos hb-topos error-topos bp-topos)
store (Mockito/mock BlobStore)]
(.thenReturn (Mockito/when (.storedTopoIds store)) #{})
- (is (= (Nimbus/topoIdsToClean mock-state store)
+ (is (= (Nimbus/topoIdsToClean mock-state store (new java.util.HashMap))
#{}))))
(deftest do-cleanup-removes-inactive-znodes
@@ -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}]
+ conf {NIMBUS-MONITOR-FREQ-SECS 10 NIMBUS-TOPOLOGY-BLOBSTORE-DELETION-DELAY-MSEC 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))]
@@ -1920,6 +1920,7 @@
;; remove topos from heartbeat cache
(is (= (count (.get (.getHeartbeatsCache nimbus))) 0)))))))
+
(deftest do-cleanup-does-not-teardown-active-topos
(let [inactive-topos ()
hb-cache {"topo1" nil "topo2" nil}
@@ -1988,3 +1989,4 @@
(is (= (list "topo1" "authorized") supervisor-topologies))
(is (= #{"authorized"} user-topologies)))))
+
http://git-wip-us.apache.org/repos/asf/storm/blob/99f47570/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 2487414..3f04d83 100644
--- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
+++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
@@ -280,6 +280,12 @@ public class DaemonConfig implements Validated {
public static final String NIMBUS_WORKER_HEARTBEATS_RECOVERY_STRATEGY_CLASS = "nimbus.worker.heartbeats.recovery.strategy.class";
/**
+ * 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";
+
+ /**
* Storm UI binds to this host/interface.
*/
@isString
http://git-wip-us.apache.org/repos/asf/storm/blob/99f47570/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 48e8c21..9e172d0 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
@@ -187,6 +187,7 @@ import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.LocalState;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.ReflectionUtils;
+import org.apache.storm.utils.RotatingMap;
import org.apache.storm.utils.ServerConfigUtils;
import org.apache.storm.utils.ServerUtils;
import org.apache.storm.utils.SimpleVersion;
@@ -371,6 +372,8 @@ 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 long topologyCleanupRotationTime = 0L;
// END TOPOLOGY STATE TRANSITIONS
@@ -828,12 +831,54 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
}
}
+ private static void rotateTopologyCleanupMap(long deletionDelay) {
+ if (Time.currentTimeMillis() - topologyCleanupRotationTime > deletionDelay) {
+ topologyCleanupDetected.rotate();
+ topologyCleanupRotationTime = Time.currentTimeMillis();
+ }
+ }
+
+ private static long getTopologyCleanupDetectedTime(String topologyId) {
+ Long firstDetectedForDeletion = topologyCleanupDetected.get(topologyId);
+ if (firstDetectedForDeletion == null) {
+ firstDetectedForDeletion = Time.currentTimeMillis();
+ topologyCleanupDetected.put(topologyId, firstDetectedForDeletion);
+ }
+ return firstDetectedForDeletion;
+ }
+
+ /**
+ * 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.
+ *
+ * @param store blobstore to search
+ * @param conf the nimbus conf
+ * @return a set of
+ */
+ static Set<String> getIdleTopologyIds(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);
+ for (String topologyId : toposToClean) {
+ if (Time.currentTimeMillis() - getTopologyCleanupDetectedTime(topologyId) >= topologyDeletionDelay) {
+ idleTopologies.add(topologyId);
+ }
+ }
+
+ rotateTopologyCleanupMap(topologyDeletionDelay);
+
+ return idleTopologies;
+ }
+
@VisibleForTesting
- public static Set<String> topoIdsToClean(IStormClusterState state, BlobStore store) {
+ public static Set<String> topoIdsToClean(IStormClusterState state, BlobStore store, Map<String, Object> conf) {
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(store.storedTopoIds(), EMPTY_STRING_SET));
+ ret.addAll(Utils.OR(getIdleTopologyIds(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));
@@ -2377,7 +2422,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
IStormClusterState state = stormClusterState;
Set<String> toClean;
synchronized (submitLock) {
- toClean = topoIdsToClean(state, blobStore);
+ toClean = topoIdsToClean(state, blobStore, this.conf);
}
if (toClean != null) {
for (String topoId : toClean) {
@@ -3306,7 +3351,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
try {
String sessionId = Utils.uuid();
blobUploaders.put(sessionId, blobStore.createBlob(key, meta, getSubject()));
- LOG.info("Created blob for {}", key);
+ LOG.info("Created blob {} for session {}", key, sessionId);
return sessionId;
} catch (Exception e) {
LOG.warn("begin create blob exception.", e);
@@ -3437,7 +3482,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
(int) conf
.getOrDefault(Config.STORM_BLOBSTORE_INPUTSTREAM_BUFFER_SIZE_BYTES,
65536)));
- LOG.info("Created download session for {}", key);
+ LOG.info("Created download session {} for {}", sessionId, key);
return ret;
} catch (Exception e) {
LOG.warn("begin blob download exception.", e);
http://git-wip-us.apache.org/repos/asf/storm/blob/99f47570/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 47484ac..bad75f0 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
@@ -18,14 +18,23 @@
package org.apache.storm.daemon.nimbus;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
+import org.apache.storm.blobstore.BlobStore;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.scheduler.resource.strategies.priority.DefaultSchedulingPriorityStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy;
import org.apache.storm.testing.TestWordSpout;
import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.utils.Time;
+import org.junit.Assert;
import org.junit.Test;
+import org.mockito.Mockito;
import static org.junit.Assert.fail;
@@ -55,4 +64,25 @@ public class NimbusTest {
//Expected...
}
}
+
+ @Test
+ public void uploadedBlobPersistsMinimumTime() {
+ BlobStore store = Mockito.mock(BlobStore.class);
+ Set<String> idleTopologies = new HashSet<>();
+ idleTopologies.add("topology1");
+ Mockito.when(store.storedTopoIds()).thenReturn(idleTopologies);
+ Map<String, Object> conf = new HashMap<>();
+
+ try (Time.SimulatedTime t = new Time.SimulatedTime(null)) {
+ Set<String> toDelete = Nimbus.getIdleTopologyIds(store, conf);
+ Assert.assertTrue(toDelete.isEmpty());
+
+ Time.advanceTime(10 * 60 * 1000L);
+
+ toDelete = Nimbus.getIdleTopologyIds(store, conf);
+ Assert.assertTrue(toDelete.contains("topology1"));
+ Assert.assertEquals(1, toDelete.size());
+
+ }
+ }
}
[2/3] storm git commit: review rework
Posted by sr...@apache.org.
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());
[3/3] storm git commit: Merge branch 'agresch_storm3053' of
https://github.com/agresch/storm into STORM-3053-merge
Posted by sr...@apache.org.
Merge branch 'agresch_storm3053' of https://github.com/agresch/storm into STORM-3053-merge
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/b0bdbf08
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/b0bdbf08
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/b0bdbf08
Branch: refs/heads/master
Commit: b0bdbf08260d55d36a5893cf3dcafba90d695211
Parents: 0bc4e26 787a434
Author: Stig Rohde Døssing <sr...@apache.org>
Authored: Fri Jun 1 17:16:40 2018 +0200
Committer: Stig Rohde Døssing <sr...@apache.org>
Committed: Fri Jun 1 17:16:40 2018 +0200
----------------------------------------------------------------------
conf/defaults.yaml | 2 +
.../test/clj/org/apache/storm/nimbus_test.clj | 10 ++--
.../java/org/apache/storm/DaemonConfig.java | 6 +++
.../org/apache/storm/daemon/nimbus/Nimbus.java | 57 ++++++++++++++++++--
.../apache/storm/daemon/nimbus/NimbusTest.java | 31 +++++++++++
5 files changed, 97 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/b0bdbf08/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
----------------------------------------------------------------------