This is an automated email from the ASF dual-hosted git repository.
ericpai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new f3c61e8543 [IOTDB-3958] Add error logging in thread pool (#6772)
f3c61e8543 is described below
commit f3c61e854366b4e1030b7211279ee76c96321087
Author: BaiJian <[email protected]>
AuthorDate: Wed Jul 27 10:47:02 2022 +0800
[IOTDB-3958] Add error logging in thread pool (#6772)
---
.../{WrappedRunnable.java => WrappedCallable.java} | 32 ++++++++++++----------
.../iotdb/commons/concurrent/WrappedRunnable.java | 24 ++++++++--------
.../threadpool/ScheduledExecutorUtil.java | 7 +++++
.../WrappedScheduledExecutorService.java | 31 +++++++++++++--------
.../WrappedSingleThreadExecutorService.java | 23 ++++++++++------
.../WrappedSingleThreadScheduledExecutor.java | 31 +++++++++++++--------
.../threadpool/WrappedThreadPoolExecutor.java | 25 +++++++++++++++++
7 files changed, 116 insertions(+), 57 deletions(-)
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedRunnable.java
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedCallable.java
similarity index 56%
copy from
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedRunnable.java
copy to
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedCallable.java
index 738c31b785..ae42a0dc4f 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedRunnable.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedCallable.java
@@ -18,29 +18,33 @@
*/
package org.apache.iotdb.commons.concurrent;
-import com.google.common.base.Throwables;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
-public abstract class WrappedRunnable implements Runnable {
+import java.util.concurrent.Callable;
- private static final Logger LOGGER =
LoggerFactory.getLogger(WrappedRunnable.class);
+/** A wrapper for {@link Callable} logging errors when uncaught exception is
thrown. */
+public abstract class WrappedCallable<V> implements Callable<V> {
@Override
- public final void run() {
+ public final V call() {
try {
- runMayThrow();
+ return callMayThrow();
} catch (Exception e) {
- LOGGER.error(e.getMessage(), e);
- throw propagate(e);
+ throw ScheduledExecutorUtil.propagate(e);
}
}
- public abstract void runMayThrow() throws Exception;
+ public abstract V callMayThrow() throws Exception;
- @SuppressWarnings("squid:S112")
- private static RuntimeException propagate(Throwable throwable) {
- Throwables.throwIfUnchecked(throwable);
- throw new RuntimeException(throwable);
+ public static <V> Callable<V> wrap(Callable<V> callable) {
+ if (callable instanceof WrappedCallable) {
+ return callable;
+ }
+ return new WrappedCallable<V>() {
+ @Override
+ public V callMayThrow() throws Exception {
+ return callable.call();
+ }
+ };
}
}
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedRunnable.java
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedRunnable.java
index 738c31b785..a91278639d 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedRunnable.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedRunnable.java
@@ -18,29 +18,31 @@
*/
package org.apache.iotdb.commons.concurrent;
-import com.google.common.base.Throwables;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+/** A wrapper for {@link Runnable} logging errors when uncaught exception is
thrown. */
public abstract class WrappedRunnable implements Runnable {
- private static final Logger LOGGER =
LoggerFactory.getLogger(WrappedRunnable.class);
-
@Override
public final void run() {
try {
runMayThrow();
} catch (Exception e) {
- LOGGER.error(e.getMessage(), e);
- throw propagate(e);
+ throw ScheduledExecutorUtil.propagate(e);
}
}
public abstract void runMayThrow() throws Exception;
- @SuppressWarnings("squid:S112")
- private static RuntimeException propagate(Throwable throwable) {
- Throwables.throwIfUnchecked(throwable);
- throw new RuntimeException(throwable);
+ public static Runnable wrap(Runnable runnable) {
+ if (runnable instanceof WrappedRunnable) {
+ return runnable;
+ }
+ return new WrappedRunnable() {
+ @Override
+ public void runMayThrow() {
+ runnable.run();
+ }
+ };
}
}
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/ScheduledExecutorUtil.java
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/ScheduledExecutorUtil.java
index 1ed2993a12..36b4f189ad 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/ScheduledExecutorUtil.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/ScheduledExecutorUtil.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.commons.concurrent.threadpool;
+import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -185,4 +186,10 @@ public class ScheduledExecutorUtil {
delay,
unit);
}
+
+ public static RuntimeException propagate(Throwable throwable) {
+ logger.error("Run thread failed", throwable);
+ Throwables.throwIfUnchecked(throwable);
+ throw new RuntimeException(throwable);
+ }
}
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedScheduledExecutorService.java
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedScheduledExecutorService.java
index a362983785..1dbd7f369c 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedScheduledExecutorService.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedScheduledExecutorService.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.commons.concurrent.threadpool;
+import org.apache.iotdb.commons.concurrent.WrappedCallable;
+import org.apache.iotdb.commons.concurrent.WrappedRunnable;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.service.JMXService;
@@ -33,6 +35,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
public class WrappedScheduledExecutorService
implements ScheduledExecutorService, WrappedScheduledExecutorServiceMBean {
@@ -49,26 +52,26 @@ public class WrappedScheduledExecutorService
@Override
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit
unit) {
- return service.schedule(command, delay, unit);
+ return service.schedule(WrappedRunnable.wrap(command), delay, unit);
}
@Override
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay,
TimeUnit unit) {
- return service.schedule(callable, delay, unit);
+ return service.schedule(WrappedCallable.wrap(callable), delay, unit);
}
@Override
@SuppressWarnings("unsafeThreadSchedule")
public ScheduledFuture<?> scheduleAtFixedRate(
Runnable command, long initialDelay, long period, TimeUnit unit) {
- return service.scheduleAtFixedRate(command, initialDelay, period, unit);
+ return service.scheduleAtFixedRate(WrappedRunnable.wrap(command),
initialDelay, period, unit);
}
@Override
@SuppressWarnings("unsafeThreadSchedule")
public ScheduledFuture<?> scheduleWithFixedDelay(
Runnable command, long initialDelay, long delay, TimeUnit unit) {
- return service.scheduleWithFixedDelay(command, initialDelay, delay, unit);
+ return service.scheduleWithFixedDelay(WrappedRunnable.wrap(command),
initialDelay, delay, unit);
}
@Override
@@ -100,47 +103,51 @@ public class WrappedScheduledExecutorService
@Override
public <T> Future<T> submit(Callable<T> task) {
- return service.submit(task);
+ return service.submit(WrappedCallable.wrap(task));
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
- return service.submit(task, result);
+ return service.submit(WrappedRunnable.wrap(task), result);
}
@Override
public Future<?> submit(Runnable task) {
- return service.submit(task);
+ return service.submit(WrappedRunnable.wrap(task));
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
- return service.invokeAll(tasks);
+ return service.invokeAll(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
}
@Override
public <T> List<Future<T>> invokeAll(
Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
- return service.invokeAll(tasks, timeout, unit);
+ return service.invokeAll(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
- return service.invokeAny(tasks);
+ return service.invokeAny(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long
timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
- return service.invokeAny(tasks, timeout, unit);
+ return service.invokeAny(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
}
@Override
public void execute(Runnable command) {
- service.execute(command);
+ service.execute(WrappedRunnable.wrap(command));
}
@Override
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java
index 9035c4ba35..f597669815 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.commons.concurrent.threadpool;
+import org.apache.iotdb.commons.concurrent.WrappedCallable;
+import org.apache.iotdb.commons.concurrent.WrappedRunnable;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.service.JMXService;
@@ -30,6 +32,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
public class WrappedSingleThreadExecutorService
implements ExecutorService, WrappedSingleThreadExecutorServiceMBean {
@@ -74,46 +77,50 @@ public class WrappedSingleThreadExecutorService
@Override
public <T> Future<T> submit(Callable<T> task) {
- return service.submit(task);
+ return service.submit(WrappedCallable.wrap(task));
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
- return service.submit(task, result);
+ return service.submit(WrappedRunnable.wrap(task), result);
}
@Override
public Future<?> submit(Runnable task) {
- return service.submit(task);
+ return service.submit(WrappedRunnable.wrap(task));
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
- return service.invokeAll(tasks);
+ return service.invokeAll(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
}
@Override
public <T> List<Future<T>> invokeAll(
Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
- return service.invokeAll(tasks, timeout, unit);
+ return service.invokeAll(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
- return service.invokeAny(tasks);
+ return service.invokeAny(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long
timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
- return service.invokeAny(tasks, timeout, unit);
+ return service.invokeAny(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
}
@Override
public void execute(Runnable command) {
- service.execute(command);
+ service.execute(WrappedRunnable.wrap(command));
}
}
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java
index 40d5e7183c..b8238a0405 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.commons.concurrent.threadpool;
+import org.apache.iotdb.commons.concurrent.WrappedCallable;
+import org.apache.iotdb.commons.concurrent.WrappedRunnable;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.service.JMXService;
@@ -31,6 +33,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
public class WrappedSingleThreadScheduledExecutor
implements ScheduledExecutorService,
WrappedSingleThreadScheduledExecutorMBean {
@@ -47,26 +50,26 @@ public class WrappedSingleThreadScheduledExecutor
@Override
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit
unit) {
- return service.schedule(command, delay, unit);
+ return service.schedule(WrappedRunnable.wrap(command), delay, unit);
}
@Override
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay,
TimeUnit unit) {
- return service.schedule(callable, delay, unit);
+ return service.schedule(WrappedCallable.wrap(callable), delay, unit);
}
@Override
@SuppressWarnings("unsafeThreadSchedule")
public ScheduledFuture<?> scheduleAtFixedRate(
Runnable command, long initialDelay, long period, TimeUnit unit) {
- return service.scheduleAtFixedRate(command, initialDelay, period, unit);
+ return service.scheduleAtFixedRate(WrappedRunnable.wrap(command),
initialDelay, period, unit);
}
@Override
@SuppressWarnings("unsafeThreadSchedule")
public ScheduledFuture<?> scheduleWithFixedDelay(
Runnable command, long initialDelay, long delay, TimeUnit unit) {
- return service.scheduleWithFixedDelay(command, initialDelay, delay, unit);
+ return service.scheduleWithFixedDelay(WrappedRunnable.wrap(command),
initialDelay, delay, unit);
}
@Override
@@ -98,46 +101,50 @@ public class WrappedSingleThreadScheduledExecutor
@Override
public <T> Future<T> submit(Callable<T> task) {
- return service.submit(task);
+ return service.submit(WrappedCallable.wrap(task));
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
- return service.submit(task, result);
+ return service.submit(WrappedRunnable.wrap(task), result);
}
@Override
public Future<?> submit(Runnable task) {
- return service.submit(task);
+ return service.submit(WrappedRunnable.wrap(task));
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
- return service.invokeAll(tasks);
+ return service.invokeAll(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
}
@Override
public <T> List<Future<T>> invokeAll(
Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
- return service.invokeAll(tasks, timeout, unit);
+ return service.invokeAll(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
- return service.invokeAny(tasks);
+ return service.invokeAny(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long
timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
- return service.invokeAny(tasks, timeout, unit);
+ return service.invokeAny(
+
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
}
@Override
public void execute(Runnable command) {
- service.execute(command);
+ service.execute(WrappedRunnable.wrap(command));
}
}
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedThreadPoolExecutor.java
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedThreadPoolExecutor.java
index 834f5d5500..4538603e60 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedThreadPoolExecutor.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedThreadPoolExecutor.java
@@ -23,14 +23,21 @@ import org.apache.iotdb.commons.concurrent.IoTThreadFactory;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.service.JMXService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.util.List;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class WrappedThreadPoolExecutor extends ThreadPoolExecutor
implements WrappedThreadPoolExecutorMBean {
+ private static final Logger logger =
LoggerFactory.getLogger(WrappedThreadPoolExecutor.class);
private final String mbeanName;
public WrappedThreadPoolExecutor(
@@ -79,4 +86,22 @@ public class WrappedThreadPoolExecutor extends
ThreadPoolExecutor
public int getQueueLength() {
return getQueue().size();
}
+
+ protected void afterExecute(Runnable r, Throwable t) {
+ super.afterExecute(r, t);
+ if (t == null && r instanceof Future<?>) {
+ try {
+ ((Future<?>) r).get();
+ } catch (CancellationException ce) {
+ t = ce;
+ } catch (ExecutionException ee) {
+ t = ee.getCause();
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ if (t != null) {
+ logger.error("Exception in thread pool {}", mbeanName, t);
+ }
+ }
}
