You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2018/10/09 20:07:38 UTC
[1/2] storm git commit: STORM-3249: Make sure times shut down and so
does shut down thread
Repository: storm
Updated Branches:
refs/heads/master a0f3b1ace -> 22440747a
STORM-3249: Make sure times shut down and so does shut down thread
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/bbfd7cd4
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/bbfd7cd4
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/bbfd7cd4
Branch: refs/heads/master
Commit: bbfd7cd4050ae80d406daaed9cec882b00be1f98
Parents: 20d5581
Author: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Authored: Mon Oct 8 13:25:17 2018 -0500
Committer: Robert (Bobby) Evans <ev...@yahoo-inc.com>
Committed: Mon Oct 8 13:25:17 2018 -0500
----------------------------------------------------------------------
.../org/apache/storm/hdfs/spout/HdfsSpout.java | 2 +-
.../org/apache/storm/hive/bolt/HiveBolt.java | 5 ++-
.../apache/storm/hive/trident/HiveState.java | 3 +-
.../src/jvm/org/apache/storm/utils/Utils.java | 43 ++++++++++++--------
.../storm/blobstore/LocalFsBlobStore.java | 5 ++-
.../java/org/apache/storm/daemon/drpc/DRPC.java | 6 +--
6 files changed, 40 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/bbfd7cd4/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java
index 12debe5..bdf9da3 100644
--- a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java
+++ b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java
@@ -390,7 +390,7 @@ public class HdfsSpout extends BaseRichSpout {
public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
LOG.info("Opening HDFS Spout");
this.conf = conf;
- this.commitTimer = new Timer();
+ this.commitTimer = new Timer(context.getThisTaskId() + "-commit-timer", true);
this.tracker = new ProgressTracker();
this.hdfsConfig = new Configuration();
this.collector = collector;
http://git-wip-us.apache.org/repos/asf/storm/blob/bbfd7cd4/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/HiveBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/HiveBolt.java b/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/HiveBolt.java
index cfabbd6..180f41b 100644
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/HiveBolt.java
+++ b/external/storm-hive/src/main/java/org/apache/storm/hive/bolt/HiveBolt.java
@@ -79,7 +79,7 @@ public class HiveBolt extends BaseRichBolt {
new ThreadFactoryBuilder().setNameFormat(timeoutName).build());
sendHeartBeat.set(true);
- heartBeatTimer = new Timer();
+ heartBeatTimer = new Timer(topologyContext.getThisTaskId() + "-hb-timer", true);
setupHeartBeatTimer();
} catch (Exception e) {
@@ -151,6 +151,9 @@ public class HiveBolt extends BaseRichBolt {
}
callTimeoutPool = null;
+ if (heartBeatTimer != null) {
+ heartBeatTimer.cancel();
+ }
super.cleanup();
LOG.info("Hive Bolt stopped");
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bbfd7cd4/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveState.java
----------------------------------------------------------------------
diff --git a/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveState.java b/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveState.java
index a698e24..6717329 100644
--- a/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveState.java
+++ b/external/storm-hive/src/main/java/org/apache/storm/hive/trident/HiveState.java
@@ -80,7 +80,7 @@ public class HiveState implements State {
String timeoutName = "hive-bolt-%d";
this.callTimeoutPool = Executors.newFixedThreadPool(1,
new ThreadFactoryBuilder().setNameFormat(timeoutName).build());
- heartBeatTimer = new Timer();
+ heartBeatTimer = new Timer("hive-hb-timer", true);
setupHeartBeatTimer();
} catch (Exception e) {
LOG.warn("unable to make connection to hive ", e);
@@ -289,6 +289,7 @@ public class HiveState implements State {
LOG.warn("shutdown interrupted on " + execService, ex);
}
}
+ heartBeatTimer.cancel();
callTimeoutPool = null;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bbfd7cd4/storm-client/src/jvm/org/apache/storm/utils/Utils.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/utils/Utils.java b/storm-client/src/jvm/org/apache/storm/utils/Utils.java
index a0406e0..9e9196f 100644
--- a/storm-client/src/jvm/org/apache/storm/utils/Utils.java
+++ b/storm-client/src/jvm/org/apache/storm/utils/Utils.java
@@ -33,6 +33,7 @@ import java.io.ObjectOutputStream;
import java.io.OutputStreamWriter;
import java.io.Serializable;
import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.URL;
@@ -300,21 +301,25 @@ public class Utils {
* runtime to avoid any zombie process in case cleanup function hangs.
*/
public static void addShutdownHookWithDelayedForceKill(Runnable func, int numSecs) {
- Runnable sleepKill = new Runnable() {
- @Override
- public void run() {
- try {
- LOG.info("Halting after {} seconds", numSecs);
- Time.sleepSecs(numSecs);
- LOG.warn("Forcing Halt...");
- Runtime.getRuntime().halt(20);
- } catch (Exception e) {
- LOG.warn("Exception in the ShutDownHook", e);
- }
+ final Thread sleepKill = new Thread(() -> {
+ try {
+ LOG.info("Halting after {} seconds", numSecs);
+ Time.sleepSecs(numSecs);
+ LOG.warn("Forcing Halt... {}", Utils.threadDump());
+ Runtime.getRuntime().halt(20);
+ } catch (InterruptedException ie) {
+ //Ignored/expected...
+ } catch (Exception e) {
+ LOG.warn("Exception in the ShutDownHook", e);
}
- };
- Runtime.getRuntime().addShutdownHook(new Thread(func));
- Runtime.getRuntime().addShutdownHook(new Thread(sleepKill));
+ });
+ sleepKill.setDaemon(true);
+ Thread wrappedFunc = new Thread(() -> {
+ func.run();
+ sleepKill.interrupt();
+ });
+ Runtime.getRuntime().addShutdownHook(wrappedFunc);
+ Runtime.getRuntime().addShutdownHook(sleepKill);
}
public static boolean isSystemId(String id) {
@@ -1190,7 +1195,9 @@ public class Utils {
final StringBuilder dump = new StringBuilder();
final java.lang.management.ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
final java.lang.management.ThreadInfo[] threadInfos = threadMXBean.getThreadInfo(threadMXBean.getAllThreadIds(), 100);
- for (java.lang.management.ThreadInfo threadInfo : threadInfos) {
+ for (Entry<Thread, StackTraceElement[]> entry: Thread.getAllStackTraces().entrySet()) {
+ Thread t = entry.getKey();
+ ThreadInfo threadInfo = threadMXBean.getThreadInfo(t.getId());
if (threadInfo == null) {
//Thread died before we could get the info, skip
continue;
@@ -1198,6 +1205,9 @@ public class Utils {
dump.append('"');
dump.append(threadInfo.getThreadName());
dump.append("\" ");
+ if (t.isDaemon()) {
+ dump.append("(DAEMON)");
+ }
dump.append("\n lock: ");
dump.append(threadInfo.getLockName());
dump.append(" owner: ");
@@ -1205,8 +1215,7 @@ public class Utils {
final Thread.State state = threadInfo.getThreadState();
dump.append("\n java.lang.Thread.State: ");
dump.append(state);
- final StackTraceElement[] stackTraceElements = threadInfo.getStackTrace();
- for (final StackTraceElement stackTraceElement : stackTraceElements) {
+ for (final StackTraceElement stackTraceElement : entry.getValue()) {
dump.append("\n at ");
dump.append(stackTraceElement);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bbfd7cd4/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java b/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java
index a72944b..7724b31 100644
--- a/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java
+++ b/storm-server/src/main/java/org/apache/storm/blobstore/LocalFsBlobStore.java
@@ -108,7 +108,7 @@ public class LocalFsBlobStore extends BlobStore {
} catch (Exception e) {
e.printStackTrace();
}
- timer = new Timer();
+ timer = new Timer("BLOB-STORE-TIMER", true);
this.leaderElector = leaderElector;
}
@@ -411,6 +411,9 @@ public class LocalFsBlobStore extends BlobStore {
if (zkClient != null) {
zkClient.close();
}
+ if (timer != null) {
+ timer.cancel();;
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/bbfd7cd4/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java b/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java
index e8edd7d..2d853f3 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/drpc/DRPC.java
@@ -71,7 +71,7 @@ public class DRPC implements AutoCloseable {
//Waiting to be returned
private final ConcurrentHashMap<String, OutstandingRequest> _requests =
new ConcurrentHashMap<>();
- private final Timer _timer = new Timer();
+ private final Timer timer = new Timer("DRPC-CLEANUP-TIMER", true);
private final AtomicLong _ctr = new AtomicLong(0);
private final IAuthorizer _auth;
@@ -87,7 +87,7 @@ public class DRPC implements AutoCloseable {
this.meterResultCalls = metricsRegistry.registerMeter("drpc:num-result-calls");
this.meterFailRequestCalls = metricsRegistry.registerMeter("drpc:num-failRequest-calls");
this.meterFetchRequestCalls = metricsRegistry.registerMeter("drpc:num-fetchRequest-calls");
- _timer.scheduleAtFixedRate(new TimerTask() {
+ timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
cleanupAll(timeoutMs, TIMED_OUT);
@@ -241,7 +241,7 @@ public class DRPC implements AutoCloseable {
@Override
public void close() {
- _timer.cancel();
+ timer.cancel();
cleanupAll(0, SHUT_DOWN);
}
}
[2/2] storm git commit: Merge branch 'STORM-3249' of
https://github.com/revans2/incubator-storm into STORM-3249
Posted by bo...@apache.org.
Merge branch 'STORM-3249' of https://github.com/revans2/incubator-storm into STORM-3249
STORM-3249: Make sure times shut down and so does shut down thread
This closes #2867
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/22440747
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/22440747
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/22440747
Branch: refs/heads/master
Commit: 22440747aa0f2ed3f3d26baec9140f79bb82f644
Parents: a0f3b1a bbfd7cd
Author: Robert Evans <ev...@yahoo-inc.com>
Authored: Tue Oct 9 14:35:30 2018 -0500
Committer: Robert Evans <ev...@yahoo-inc.com>
Committed: Tue Oct 9 14:35:30 2018 -0500
----------------------------------------------------------------------
.../org/apache/storm/hdfs/spout/HdfsSpout.java | 2 +-
.../org/apache/storm/hive/bolt/HiveBolt.java | 5 ++-
.../apache/storm/hive/trident/HiveState.java | 3 +-
.../src/jvm/org/apache/storm/utils/Utils.java | 43 ++++++++++++--------
.../storm/blobstore/LocalFsBlobStore.java | 5 ++-
.../java/org/apache/storm/daemon/drpc/DRPC.java | 6 +--
6 files changed, 40 insertions(+), 24 deletions(-)
----------------------------------------------------------------------