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/07/14 09:48:18 UTC
[1/3] storm git commit: STORM-1928 ShellSpout should only check
heartbeat while ShellSpout is waiting for subprocess to sync
Repository: storm
Updated Branches:
refs/heads/1.0.x-branch d29d13c0b -> c359889d9
STORM-1928 ShellSpout should only check heartbeat while ShellSpout is waiting for subprocess to sync
* only check heartbeat while ShellSpout is waiting for subprocess to sync
* change ShellSpout.die logic similar to ShellBolt
* ShellSpout's die message is logged but worker is not died after logging
* Confirmed that worker becomes died when die has been called
* rename method to more meaningful name
* mark waiting subprocess first to check heartbeat while also sending spout message to subprocess
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/25f2db76
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/25f2db76
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/25f2db76
Branch: refs/heads/1.0.x-branch
Commit: 25f2db76492b84689c32e90a256bc00bdbe8b07b
Parents: d29d13c
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Tue Jun 28 00:07:59 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Thu Jul 14 18:46:50 2016 +0900
----------------------------------------------------------------------
.../jvm/org/apache/storm/spout/ShellSpout.java | 56 ++++++++++++++++----
1 file changed, 46 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/25f2db76/storm-core/src/jvm/org/apache/storm/spout/ShellSpout.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/spout/ShellSpout.java b/storm-core/src/jvm/org/apache/storm/spout/ShellSpout.java
index e24c76c..72003f4 100644
--- a/storm-core/src/jvm/org/apache/storm/spout/ShellSpout.java
+++ b/storm-core/src/jvm/org/apache/storm/spout/ShellSpout.java
@@ -25,6 +25,8 @@ import org.apache.storm.multilang.ShellMsg;
import org.apache.storm.multilang.SpoutMsg;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.utils.ShellProcess;
+
+import java.util.Arrays;
import java.util.Map;
import java.util.HashMap;
import java.util.List;
@@ -32,6 +34,7 @@ import java.util.TimerTask;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import clojure.lang.RT;
@@ -47,7 +50,9 @@ public class ShellSpout implements ISpout {
private String[] _command;
private Map<String, String> env = new HashMap<>();
private ShellProcess _process;
-
+ private volatile boolean _running = true;
+ private volatile RuntimeException _exception;
+
private TopologyContext _context;
private SpoutMsg _spoutMsg;
@@ -55,6 +60,7 @@ public class ShellSpout implements ISpout {
private int workerTimeoutMills;
private ScheduledExecutorService heartBeatExecutorService;
private AtomicLong lastHeartbeatTimestamp = new AtomicLong();
+ private AtomicBoolean waitingOnSubprocess = new AtomicBoolean(false);
public ShellSpout(ShellComponent component) {
this(component.get_execution_command(), component.get_script());
@@ -94,9 +100,14 @@ public class ShellSpout implements ISpout {
public void close() {
heartBeatExecutorService.shutdownNow();
_process.destroy();
+ _running = false;
}
public void nextTuple() {
+ if (_exception != null) {
+ throw _exception;
+ }
+
if (_spoutMsg == null) {
_spoutMsg = new SpoutMsg();
}
@@ -106,6 +117,10 @@ public class ShellSpout implements ISpout {
}
public void ack(Object msgId) {
+ if (_exception != null) {
+ throw _exception;
+ }
+
if (_spoutMsg == null) {
_spoutMsg = new SpoutMsg();
}
@@ -115,6 +130,10 @@ public class ShellSpout implements ISpout {
}
public void fail(Object msgId) {
+ if (_exception != null) {
+ throw _exception;
+ }
+
if (_spoutMsg == null) {
_spoutMsg = new SpoutMsg();
}
@@ -153,6 +172,7 @@ public class ShellSpout implements ISpout {
private void querySubprocess() {
try {
+ markWaitingSubprocess();
_process.writeSpoutMsg(_spoutMsg);
while (true) {
@@ -192,9 +212,12 @@ public class ShellSpout implements ISpout {
} catch (Exception e) {
String processInfo = _process.getProcessInfoString() + _process.getProcessTerminationInfoString();
throw new RuntimeException(processInfo, e);
+ } finally {
+ completedWaitingSubprocess();
}
}
+
private void handleLog(ShellMsg shellMsg) {
String msg = shellMsg.getMsg();
msg = "ShellLog " + _process.getProcessInfoString() + " " + msg;
@@ -247,13 +270,25 @@ public class ShellSpout implements ISpout {
return lastHeartbeatTimestamp.get();
}
- private void die(Throwable exception) {
- heartBeatExecutorService.shutdownNow();
+ private void markWaitingSubprocess() {
+ waitingOnSubprocess.compareAndSet(false, true);
+ }
- LOG.error("Halting process: ShellSpout died.", exception);
+ private void completedWaitingSubprocess() {
+ waitingOnSubprocess.compareAndSet(true, false);
+ }
+
+ private void die(Throwable exception) {
+ String processInfo = _process.getProcessInfoString() + _process.getProcessTerminationInfoString();
+ _exception = new RuntimeException(processInfo, exception);
+ String message = String.format("Halting process: ShellSpout died. Command: %s, ProcessInfo %s",
+ Arrays.toString(_command),
+ processInfo);
+ LOG.error(message, exception);
_collector.reportError(exception);
- _process.destroy();
- System.exit(11);
+ if (_running || (exception instanceof Error)) { //don't exit if not running, unless it is an Error
+ System.exit(11);
+ }
}
private class SpoutHeartbeatTimerTask extends TimerTask {
@@ -265,13 +300,14 @@ public class ShellSpout implements ISpout {
@Override
public void run() {
- long currentTimeMillis = System.currentTimeMillis();
long lastHeartbeat = getLastHeartbeat();
+ long currentTimestamp = System.currentTimeMillis();
+ boolean isWaitingOnSubprocess = waitingOnSubprocess.get();
- LOG.debug("current time : {}, last heartbeat : {}, worker timeout (ms) : {}",
- currentTimeMillis, lastHeartbeat, workerTimeoutMills);
+ LOG.debug("last heartbeat : {}, waiting subprocess now : {}, worker timeout (ms) : {}",
+ lastHeartbeat, isWaitingOnSubprocess, workerTimeoutMills);
- if (currentTimeMillis - lastHeartbeat > workerTimeoutMills) {
+ if (isWaitingOnSubprocess && currentTimestamp - lastHeartbeat > workerTimeoutMills) {
spout.die(new RuntimeException("subprocess heartbeat timeout"));
}
}
[3/3] storm git commit: add STORM-1928 to CHANGELOG
Posted by ka...@apache.org.
add STORM-1928 to CHANGELOG
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/c359889d
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/c359889d
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/c359889d
Branch: refs/heads/1.0.x-branch
Commit: c359889d9d40e6ced02fc5153efa66e6fd0c237d
Parents: 6b49dde
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Thu Jul 14 18:48:01 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Thu Jul 14 18:48:01 2016 +0900
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/c359889d/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5062cb6..4ac7c64 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 1.0.2
+ * STORM-1928: ShellSpout should check heartbeat while ShellSpout is waiting for subprocess to sync
* STORM-1922: Supervisor summary default order by host
* STORM-1956: Disabling Backpressure by default
* STORM-1934: Fix race condition between sync-supervisor and sync-processes
[2/3] storm git commit: Merge branch 'STORM-1928-1.0.x' into
1.0.x-branch
Posted by ka...@apache.org.
Merge branch 'STORM-1928-1.0.x' into 1.0.x-branch
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/6b49dde5
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/6b49dde5
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/6b49dde5
Branch: refs/heads/1.0.x-branch
Commit: 6b49dde56ce3311523b8e8f7576c01fe3a566081
Parents: d29d13c 25f2db7
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Thu Jul 14 18:46:57 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Thu Jul 14 18:46:57 2016 +0900
----------------------------------------------------------------------
.../jvm/org/apache/storm/spout/ShellSpout.java | 56 ++++++++++++++++----
1 file changed, 46 insertions(+), 10 deletions(-)
----------------------------------------------------------------------