dasahcc commented on a change in pull request #1227:
URL: https://github.com/apache/helix/pull/1227#discussion_r469588104
##########
File path:
helix-core/src/main/java/org/apache/helix/messaging/handling/HelixTaskExecutor.java
##########
@@ -177,7 +177,7 @@ public HelixTaskExecutor(ParticipantStatusMonitor
participantStatusMonitor,
_lock = new Object();
_statusUpdateUtil = new StatusUpdateUtil();
- _timer = new Timer("HelixTaskExecutor_timer", true); // created as a
daemon timer thread to handle task timeout
+ _timer = new Timer("HelixTaskExecutor_Timer", true); // created as a
daemon timer thread to handle task timeout
Review comment:
What's the difference of this line? Let's try to minimize the diff for
reviewing.
##########
File path: helix-core/src/test/java/org/apache/helix/TestHelper.java
##########
@@ -21,6 +21,7 @@
import java.io.File;
import java.io.IOException;
+import java.lang.reflect.Array;
Review comment:
Let's minimize the diff.
##########
File path:
helix-core/src/main/java/org/apache/helix/task/TaskStateModelFactory.java
##########
@@ -97,7 +97,7 @@ public void shutdown() {
}
@VisibleForTesting
- void shutdownNow() {
+ public void shutdownNow() {
Review comment:
Is the VisibleForTesting not enough? This is not a good idea to expose
it .
##########
File path: helix-core/src/test/java/org/apache/helix/TestListenerCallback.java
##########
@@ -119,6 +119,8 @@ public void beforeClass() throws Exception {
@AfterClass
public void afterClass() throws Exception {
+ String testClassName = this.getShortClassName();
+ System.out.println("AfterClass: " + testClassName + " of
TestListenerCallback called.");
Review comment:
Will this bring lots of printout? Is this for debugging purpose?
##########
File path: helix-core/src/test/java/org/apache/helix/ThreadLeakageChecker.java
##########
@@ -0,0 +1,178 @@
+package org.apache.helix;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+
+public class ThreadLeakageChecker {
+ private static ThreadGroup getRootThreadGroup() {
+ ThreadGroup candidate = Thread.currentThread().getThreadGroup();
+ while (candidate.getParent() != null) {
+ candidate = candidate.getParent();
+ }
+ return candidate;
+ }
+
+ private static List<Thread> getAllThreads() {
+ ThreadGroup rootThreadGroup = getRootThreadGroup();
+ Thread[] threads = new Thread[32];
+ int count = rootThreadGroup.enumerate(threads);
+ while (count == threads.length) {
+ threads = new Thread[threads.length * 2];
+ count = rootThreadGroup.enumerate(threads);
+ }
+ return Arrays.asList(Arrays.copyOf(threads, count));
+ }
+
+ private static final String[] ZkServerThrdPattern =
+ {"SessionTracker", "NIOServerCxn", "SyncThread:", "ProcessThread"};
+ private static final String[] ZkSessionThrdPattern =
+ new String[]{"ZkClient-EventThread", "ZkClient-AsyncCallback",
"-EventThread", "-SendThread"};
+ private static final String[] ForkJoinThrdPattern = new
String[]{"ForkJoinPool"};
+ private static final String[] TimerThrdPattern = new String[]{"time"};
+ private static final String[] TaskStateModelThrdPattern = new
String[]{"TaskStateModel"};
+
+ private static enum ThreadCategory {
+ ZkServer("zookeeper server threads", 4, 100, ZkServerThrdPattern),
+ ZkSession("zkclient/zooKeeper session threads", 12, 12,
ZkSessionThrdPattern),
+ ForkJoin("fork join pool threads", 2, 10, ForkJoinThrdPattern),
+ Timer("timer threads", 0, 2, TimerThrdPattern),
+ TaskStateModel("TaskStateModel threads", 0, 0, TaskStateModelThrdPattern),
+ Other("Other threads", 0, 3, new String[]{""});
+
+ private String _description;
+ private List<String> _pattern;
+ private int _warningLimit;
+ private int _limit;
+
+ public String getDescription() {
+ return _description;
+ }
+
+ public Predicate<String> getMatchPred() {
+ if (this.name() != ThreadCategory.Other.name()) {
+ Predicate<String> pred = target -> {
+ for (String p : _pattern) {
+ if (target.toLowerCase().contains(p.toLowerCase())) {
+ return true;
+ }
+ }
+ return false;
+ };
+ return pred;
+ }
+
+ List<Predicate<String>> predicateList = new ArrayList<>();
+ for (ThreadCategory threadCategory : ThreadCategory.values()) {
+ if (threadCategory == ThreadCategory.Other) {
+ continue;
+ }
+ predicateList.add(threadCategory.getMatchPred());
+ }
+ Predicate<String> pred = target -> {
+ for (Predicate<String> p : predicateList) {
+ if (p.test(target)) {
+ return false;
+ }
+ }
+ return true;
+ };
+
+ return pred;
+ }
+
+ public int getWarningLimit() {
+ return _warningLimit;
+ }
+
+ public int getLimit() {
+ return _limit;
+ }
+
+ private ThreadCategory(String description, int warningLimit, int limit,
String[] patterns) {
+ _description = description;
+ _pattern = Arrays.asList(patterns);
+ _warningLimit = warningLimit;
+ _limit = limit;
+ }
+ }
+
+ public static boolean afterClassCheck(String classname) {
+ // step 1: get all active threads
+ List<Thread> threads = getAllThreads();
+ System.out.println(classname + " has active threads cnt:" +
threads.size());
+
+ // step 2: categorize threads
+ Map<String, List<Thread>> threadByName = null;
+ Map<ThreadCategory, Integer> threadByCnt = new HashMap<>();
+ Map<ThreadCategory, Set<Thread>> threadByCat = new HashMap<>();
+ try {
+ threadByName = threads.
+ stream().
+ filter(p -> p.getThreadGroup() != null &&
p.getThreadGroup().getName() != null
+ && p.getThreadGroup().getName() != "system").
Review comment:
This is a bug. I dont think it will work since in Java, String compare
uses .equal().
##########
File path: helix-core/src/test/java/org/apache/helix/common/ZkTestBase.java
##########
@@ -719,6 +722,20 @@ public void cleanupLiveInstanceOwners() {
clientMap.clear();
}
_liveInstanceOwners.clear();
+
+ // wait 2 seconds for threads to be teared down.
+ // this reduce false alert.
+ Thread.sleep(1000);
+ boolean status = false;
+ try {
+ status = ThreadLeakageChecker.afterClassCheck(testClassName);
+ } catch (Exception e) {
+ System.out.println("ThreadLeakageChecker exception:" +
e.getStackTrace());
+ }
+ // Assert here does not work.
+ if (!status) {
+ System.out.println("---------- Test Class " + testClassName + " thread
leakage detected! ---------------");
Review comment:
Can we kill the threads and print log? At least, we keep the test
working.
##########
File path: helix-core/src/test/java/org/apache/helix/common/ZkTestBase.java
##########
@@ -719,6 +722,20 @@ public void cleanupLiveInstanceOwners() {
clientMap.clear();
}
_liveInstanceOwners.clear();
+
+ // wait 2 seconds for threads to be teared down.
+ // this reduce false alert.
+ Thread.sleep(1000);
Review comment:
Are we still using thread.sleep? At least, we should have periodical
check, for example check every 10 milisecond.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
