Merge branch 'master' of github.com:ebean-orm/ebean

ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java

@@ -1,5 +1,6 @@
 package io.ebeaninternal.server.core;
 
+import io.ebean.config.BackgroundExecutorWrapper;
 import io.ebean.config.ContainerConfig;
 import io.ebean.config.DatabaseConfig;
 import io.ebean.config.DatabaseConfigProvider;
@@ -67,7 +68,8 @@ public final class DefaultContainer implements SpiContainer {
     String namePrefix = "ebean-" + config.getName();
     int schedulePoolSize = config.getBackgroundExecutorSchedulePoolSize();
     int shutdownSecs = config.getBackgroundExecutorShutdownSecs();
-    return new DefaultBackgroundExecutor(schedulePoolSize, shutdownSecs, namePrefix);
+    BackgroundExecutorWrapper wrapper = config.getBackgroundExecutorWrapper();
+    return new DefaultBackgroundExecutor(schedulePoolSize, shutdownSecs, namePrefix, wrapper);
   }
 
   /**

ebean-core/src/main/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutor.java

@@ -1,70 +1,94 @@
 package io.ebeaninternal.server.executor;
 
 import io.avaje.lang.NonNullApi;
+import io.ebean.config.BackgroundExecutorWrapper;
 import io.ebeaninternal.api.SpiBackgroundExecutor;
-import org.slf4j.MDC;
 
-import java.util.Map;
 import java.util.concurrent.*;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * The default implementation of the BackgroundExecutor.
  */
 @NonNullApi
 public final class DefaultBackgroundExecutor implements SpiBackgroundExecutor {
 
+  protected static final Logger logger = LoggerFactory.getLogger("io.ebean.BackgroundExecutor");
+  
   private final ScheduledExecutorService schedulePool;
   private final DaemonExecutorService pool;
+  private final BackgroundExecutorWrapper wrapper;
 
   /**
    * Construct the default implementation of BackgroundExecutor.
    */
-  public DefaultBackgroundExecutor(int schedulePoolSize, int shutdownWaitSeconds, String namePrefix) {
+  public DefaultBackgroundExecutor(int schedulePoolSize, int shutdownWaitSeconds, String namePrefix, BackgroundExecutorWrapper wrapper) {
     this.schedulePool = new DaemonScheduleThreadPool(schedulePoolSize, shutdownWaitSeconds, namePrefix + "-periodic-");
     this.pool = new DaemonExecutorService(shutdownWaitSeconds, namePrefix);
+    this.wrapper = wrapper;
+    logger.debug("Created backgroundExecutor {} (schedulePoolSize={}, shutdownWaitSeconds={})", namePrefix, schedulePoolSize, shutdownWaitSeconds);
   }
 
   /**
    * Wrap the task with MDC context if defined.
    */
-  <T> Callable<T> wrapMDC(Callable<T> task) {
-    final Map<String, String> map = MDC.getCopyOfContextMap();
-    if (map == null) {
+  <T> Callable<T> wrap(Callable<T> task) {
+    if (wrapper == null) {
       return task;
     } else {
-      return () -> {
-        MDC.setContextMap(map);
-        try {
-          return task.call();
-        } finally {
-          MDC.clear();
-        }
-      };
+      return wrapper.wrap(task);
     }
   }
 
   /**
    * Wrap the task with MDC context if defined.
    */
-  Runnable wrapMDC(Runnable task) {
-    final Map<String, String> map = MDC.getCopyOfContextMap();
-    if (map == null) {
+  Runnable wrap(Runnable task) {
+    if (wrapper == null) {
       return task;
     } else {
-      return () -> {
-        MDC.setContextMap(map);
-        try {
+      return wrapper.wrap(task);
+    }
+  }
+  
+
+  /**
+   * Decorates a runnable by adding an exception handler and some timing metrics.
+   * This is used in methods that accepts a <code>Runnable</code> and return
+   * either <code>void</code> or <code>ScheduledFuture</code>, as there is
+   * normally no Future.get() call.
+   * 
+   * Note: When submitting a <code>Callable</code>, you must check
+   * <code>Future.get()</code> for exceptions.
+   */
+  private Runnable logExceptions(Runnable task) {
+    long queued = System.nanoTime();
+    logger.trace("Queued {}", task);
+    return () -> {
+      try {
+        if (logger.isTraceEnabled()) {
+          long start = System.nanoTime();
+          logger.trace("Start {} (delay time {} us)", task, (start - queued) / 1000L);
+          task.run();
+          logger.trace("Stop {} (exec time {} us)", task, (System.nanoTime() - start) / 1000L);
+        } else {
           task.run();
-        } finally {
-          MDC.clear();
         }
-      };
-    }
+      } catch (Throwable t) {
+        // log any exception here. Note they will not bubble up to the calling user
+        // unless Future.get() is checked. (Which is almost never done on scheduled
+        // background executions)
+        logger.error("Error while executing the task {}", task, t);
+        throw t;
+      }
+    };
   }
 
   @Override
   public <T> Future<T> submit(Callable<T> task) {
-    return pool.submit(wrapMDC(task));
+    return pool.submit(wrap(task));
   }
 
   /**
@@ -72,48 +96,52 @@ public final class DefaultBackgroundExecutor implements SpiBackgroundExecutor {
    */
   @Override
   public Future<?> submit(Runnable task) {
-    return pool.submit(wrapMDC(task));
+    return pool.submit(wrap(task));
   }
 
+  
   @Override
   public void execute(Runnable task) {
-    submit(task);
+    submit(logExceptions(task));
   }
 
   @Override
   public void executePeriodically(Runnable task, long delay, TimeUnit unit) {
-    schedulePool.scheduleWithFixedDelay(wrapMDC(task), delay, delay, unit);
+    schedulePool.scheduleWithFixedDelay(wrap(logExceptions(task)), delay, delay, unit);
   }
 
   @Override
   public void executePeriodically(Runnable task, long initialDelay, long delay, TimeUnit unit) {
-    schedulePool.scheduleWithFixedDelay(wrapMDC(task), initialDelay, delay, unit);
+    schedulePool.scheduleWithFixedDelay(wrap(logExceptions(task)), initialDelay, delay, unit);
   }
 
   @Override
   public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long initialDelay, long delay, TimeUnit unit) {
-    return schedulePool.scheduleWithFixedDelay(wrapMDC(task), initialDelay, delay, unit);
+    return schedulePool.scheduleWithFixedDelay(wrap(logExceptions(task)), initialDelay, delay, unit);
   }
 
   @Override
   public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long initialDelay, long delay, TimeUnit unit) {
-    return schedulePool.scheduleAtFixedRate(wrapMDC(task), initialDelay, delay, unit);
+    return schedulePool.scheduleAtFixedRate(wrap(logExceptions(task)), initialDelay, delay, unit);
   }
 
   @Override
   public ScheduledFuture<?> schedule(Runnable task, long delay, TimeUnit unit) {
-    return schedulePool.schedule(wrapMDC(task), delay, unit);
+    return schedulePool.schedule(wrap(logExceptions(task)), delay, unit);
   }
 
   @Override
   public <V> ScheduledFuture<V> schedule(Callable<V> task, long delay, TimeUnit unit) {
-    return schedulePool.schedule(wrapMDC(task), delay, unit);
+    // Note: Here is no "logExceptions", because it is intended to check Future.get() by the invoker
+    return schedulePool.schedule(wrap(task), delay, unit);
   }
 
   @Override
   public void shutdown() {
+    logger.trace("Shutting down backgroundExecutor");
     schedulePool.shutdown();
     pool.shutdown();
+    logger.debug("BackgroundExecutor stopped");
   }
 
 }

ebean-core/src/test/java/io/ebeaninternal/server/executor/DaemonExecutorServiceTest.java

@@ -31,7 +31,7 @@ class DaemonExecutorServiceTest {
 
   @Test
   void submit_via_DefaultBackgroundExecutor() throws Exception {
-    DefaultBackgroundExecutor des = new DefaultBackgroundExecutor(1, 5, "junk");
+    DefaultBackgroundExecutor des = new DefaultBackgroundExecutor(1, 5, "junk", null);
     long start = System.currentTimeMillis();
     List<Future<?>> futures = new ArrayList<>();
     for (int i = 0; i < count; i++) {