You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2016/11/03 00:21:00 UTC
[3/7] storm git commit: Remove time limit on Slot shutdown when
running in local mode
Remove time limit on Slot shutdown when running in local mode
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/83ea760e
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/83ea760e
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/83ea760e
Branch: refs/heads/1.x-branch
Commit: 83ea760e8c448a7630ef4d1c4f85858b8a825a1b
Parents: 9c5d4e4
Author: Stig Rohde D�ssing <st...@gmail.com>
Authored: Mon Oct 31 23:57:26 2016 +0100
Committer: Stig Rohde D�ssing <st...@gmail.com>
Committed: Tue Nov 1 00:39:01 2016 +0100
----------------------------------------------------------------------
storm-core/src/clj/org/apache/storm/testing.clj | 2 +-
storm-core/src/jvm/org/apache/storm/command/KillWorkers.java | 2 +-
.../jvm/org/apache/storm/daemon/supervisor/ReadClusterState.java | 4 ++--
.../src/jvm/org/apache/storm/daemon/supervisor/Supervisor.java | 4 ++--
.../storm/daemon/supervisor/timer/SupervisorHealthCheck.java | 2 +-
5 files changed, 7 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/83ea760e/storm-core/src/clj/org/apache/storm/testing.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/org/apache/storm/testing.clj b/storm-core/src/clj/org/apache/storm/testing.clj
index 4029be1..afb7b61 100644
--- a/storm-core/src/clj/org/apache/storm/testing.clj
+++ b/storm-core/src/clj/org/apache/storm/testing.clj
@@ -201,7 +201,7 @@
(.close (:state cluster-map))
(.disconnect (:storm-cluster-state cluster-map))
(doseq [s @(:supervisors cluster-map)]
- (.shutdownAllWorkers s)
+ (.shutdownAllWorkers s false)
;; race condition here? will it launch the workers again?
(.close s))
(ProcessSimulator/killAllProcesses)
http://git-wip-us.apache.org/repos/asf/storm/blob/83ea760e/storm-core/src/jvm/org/apache/storm/command/KillWorkers.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/command/KillWorkers.java b/storm-core/src/jvm/org/apache/storm/command/KillWorkers.java
index 1fcd81b..0ce40df 100644
--- a/storm-core/src/jvm/org/apache/storm/command/KillWorkers.java
+++ b/storm-core/src/jvm/org/apache/storm/command/KillWorkers.java
@@ -32,7 +32,7 @@ public class KillWorkers {
Map<String, Object> conf = ConfigUtils.readStormConfig();
conf.put(Config.STORM_LOCAL_DIR, new File((String)conf.get(Config.STORM_LOCAL_DIR)).getCanonicalPath());
try (Supervisor supervisor = new Supervisor(conf, null, new StandaloneSupervisor())) {
- supervisor.shutdownAllWorkers();
+ supervisor.shutdownAllWorkers(true);
}
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/83ea760e/storm-core/src/jvm/org/apache/storm/daemon/supervisor/ReadClusterState.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/ReadClusterState.java b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/ReadClusterState.java
index 7ceb691..849ba07 100644
--- a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/ReadClusterState.java
+++ b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/ReadClusterState.java
@@ -289,7 +289,7 @@ public class ReadClusterState implements Runnable, AutoCloseable {
private static final long WARN_MILLIS = 1_000; //Warn about a shutdown that takes longer than 1 second (default timeout)
private static final long ERROR_MILLIS = 10_000; //Throw an exception if after 10 seconds.
- public synchronized void shutdownAllWorkers() {
+ public synchronized void shutdownAllWorkers(boolean enableErrorOnTimeout) {
for (Slot slot: slots.values()) {
slot.setNewAssignment(null);
}
@@ -300,7 +300,7 @@ public class ReadClusterState implements Runnable, AutoCloseable {
try {
while (slot.getMachineState() != MachineState.EMPTY) {
long timeSpentMillis = Time.currentTimeMillis() - startTime;
- if (timeSpentMillis > ERROR_MILLIS) {
+ if (enableErrorOnTimeout && timeSpentMillis > ERROR_MILLIS) {
throw new IllegalStateException("It took over " + timeSpentMillis + "ms to shut down slot " + slot);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/83ea760e/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Supervisor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Supervisor.java b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Supervisor.java
index 1399e8d..5448116 100644
--- a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Supervisor.java
+++ b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/Supervisor.java
@@ -323,9 +323,9 @@ public class Supervisor implements DaemonCommon, AutoCloseable {
}
}
- public void shutdownAllWorkers() {
+ public void shutdownAllWorkers(boolean enableErrorOnTimeout) {
if (readState != null) {
- readState.shutdownAllWorkers();
+ readState.shutdownAllWorkers(enableErrorOnTimeout);
} else {
try {
ContainerLauncher launcher = ContainerLauncher.make(getConf(), getId(), getSharedContext());
http://git-wip-us.apache.org/repos/asf/storm/blob/83ea760e/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/SupervisorHealthCheck.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/SupervisorHealthCheck.java b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/SupervisorHealthCheck.java
index 0017092..a23351b 100644
--- a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/SupervisorHealthCheck.java
+++ b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/timer/SupervisorHealthCheck.java
@@ -35,7 +35,7 @@ public class SupervisorHealthCheck implements Runnable {
Map<String, Object> conf = supervisor.getConf();
int healthCode = HealthCheck.healthCheck(conf);
if (healthCode != 0) {
- supervisor.shutdownAllWorkers();
+ supervisor.shutdownAllWorkers(true);
}
}
}