Provide custom BackgroundExecutorWrapper to pass thread locals

ebean-api/src/main/java/io/ebean/config/BackgroundExecutorWrapper.java

@@ -0,0 +1,23 @@
+package io.ebean.config;
+
+import java.util.concurrent.Callable;
+
+/**
+ * BackgroundExecutorWrapper that can be used to wrap tasks that are sent to background (i.e. an other thread).
+ * It should copy all neccessary thread-local variables. See {@link MdcBackgroundExecutorWrapper} for implementation details.
+ * 
+ * @author Roland Praml, FOCONIS AG
+ */
+public interface BackgroundExecutorWrapper {
+
+  /**
+   * Wrap the task with MDC context if defined.
+   */
+  <T> Callable<T> wrap(Callable<T> task);
+
+  /**
+   * Wrap the task with MDC context if defined.
+   */
+  Runnable wrap(Runnable task);
+
+}

ebean-api/src/main/java/io/ebean/config/DatabaseConfig.java

@@ -454,6 +454,7 @@ public class DatabaseConfig {
 
   private int backgroundExecutorSchedulePoolSize = 1;
   private int backgroundExecutorShutdownSecs = 30;
+  private BackgroundExecutorWrapper backgroundExecutorWrapper = new MdcBackgroundExecutorWrapper();
 
   // defaults for the L2 bean caching
 
@@ -1480,6 +1481,20 @@ public class DatabaseConfig {
     this.backgroundExecutorShutdownSecs = backgroundExecutorShutdownSecs;
   }
 
+  /**
+   * Return the background executor wrapper.
+   */
+  public BackgroundExecutorWrapper getBackgroundExecutorWrapper() {
+    return backgroundExecutorWrapper;
+  }
+
+  /**
+   * Sets the background executor wrapper. The wrapper is used when a task is sent to background and should copy the thread-locals.
+   */
+  public void setBackgroundExecutorWrapper(BackgroundExecutorWrapper backgroundExecutorWrapper) {
+    this.backgroundExecutorWrapper = backgroundExecutorWrapper;
+  }
+
   /**
    * Return the L2 cache default max size.
    */
@@ -2926,6 +2941,7 @@ public class DatabaseConfig {
 
     backgroundExecutorSchedulePoolSize = p.getInt("backgroundExecutorSchedulePoolSize", backgroundExecutorSchedulePoolSize);
     backgroundExecutorShutdownSecs = p.getInt("backgroundExecutorShutdownSecs", backgroundExecutorShutdownSecs);
+    backgroundExecutorWrapper = p.createInstance(BackgroundExecutorWrapper.class, "backgroundExecutorWrapper", backgroundExecutorWrapper);
     disableClasspathSearch = p.getBoolean("disableClasspathSearch", disableClasspathSearch);
     currentUserProvider = p.createInstance(CurrentUserProvider.class, "currentUserProvider", currentUserProvider);
     databasePlatform = p.createInstance(DatabasePlatform.class, "databasePlatform", databasePlatform);

ebean-api/src/main/java/io/ebean/config/MdcBackgroundExecutorWrapper.java

@@ -0,0 +1,50 @@
+package io.ebean.config;
+
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import org.slf4j.MDC;
+
+public class MdcBackgroundExecutorWrapper implements BackgroundExecutorWrapper {
+
+  
+  /**
+   * Wrap the task with MDC context if defined.
+   */
+  @Override
+  public <T> Callable<T> wrap(Callable<T> task) {
+    final Map<String, String> map = MDC.getCopyOfContextMap();
+    if (map == null) {
+      return task;
+    } else {
+      return () -> {
+        MDC.setContextMap(map);
+        try {
+          return task.call();
+        } finally {
+          MDC.clear();
+        }
+      };
+    }
+  }
+
+  /**
+   * Wrap the task with MDC context if defined.
+   */
+  @Override
+  public Runnable wrap(Runnable task) {
+    final Map<String, String> map = MDC.getCopyOfContextMap();
+    if (map == null) {
+      return task;
+    } else {
+      return () -> {
+        MDC.setContextMap(map);
+        try {
+          task.run();
+        } finally {
+          MDC.clear();
+        }
+      };
+    }
+  }
+}

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,92 @@
 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) {
-      return task;
+  <T> Callable<T> wrap(Callable<T> task) {
+    if (wrapper == null) {
+      return clock(task);
     } else {
-      return () -> {
-        MDC.setContextMap(map);
-        try {
-          return task.call();
-        } finally {
-          MDC.clear();
-        }
-      };
+      return wrapper.wrap(clock(task));
     }
   }
 
   /**
    * Wrap the task with MDC context if defined.
    */
-  Runnable wrapMDC(Runnable task) {
-    final Map<String, String> map = MDC.getCopyOfContextMap();
-    if (map == null) {
-      return task;
+  Runnable wrap(Runnable task) {
+    if (wrapper == null) {
+      return clock(task);
     } else {
+      return wrapper.wrap(clock(task));
+    }
+  }
+  
+  private <T> Callable<T> clock(Callable<T> task) {
+    if (logger.isTraceEnabled()) {
+      long queued = System.nanoTime();
+      logger.trace("Queued {}", task);
       return () -> {
-        MDC.setContextMap(map);
-        try {
-          task.run();
-        } finally {
-          MDC.clear();
-        }
+        long start = System.nanoTime();
+        logger.trace("Start {} (delay time {} us)", task, (start - queued) / 1000L);
+        T ret = task.call();
+        logger.trace("Stop {} (exec time {} us)", task, (System.nanoTime() - start) / 1000L);
+        return ret;
       };
+    } else {
+      return task;
+    }
+  }
+  
+  private Runnable clock(Runnable task) {
+    if (logger.isTraceEnabled()) {
+      long queued = System.nanoTime();
+      logger.trace("Queued {}", task);
+      return () -> {
+        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 {
+      return task;
     }
   }
 
   @Override
   public <T> Future<T> submit(Callable<T> task) {
-    return pool.submit(wrapMDC(task));
+    return pool.submit(wrap(task));
   }
 
   /**
@@ -72,48 +94,56 @@ 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(() -> {
+      try {
+        task.run();
+      } catch (Throwable t) {
+        logger.error("Error while executing the task {}", task, t);
+      }
+    });
   }
 
   @Override
   public void executePeriodically(Runnable task, long delay, TimeUnit unit) {
-    schedulePool.scheduleWithFixedDelay(wrapMDC(task), delay, delay, unit);
+    schedulePool.scheduleWithFixedDelay(wrap(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(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(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(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(task), delay, unit);
   }
 
   @Override
   public <V> ScheduledFuture<V> schedule(Callable<V> task, long delay, TimeUnit unit) {
-    return schedulePool.schedule(wrapMDC(task), delay, unit);
+    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++) {

ebean-test/src/test/java/io/ebeaninternal/server/executor/DefaultBackgroundExecutorTest.java

@@ -4,17 +4,21 @@ import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.slf4j.MDC;
 
+import io.ebean.config.MdcBackgroundExecutorWrapper;
+
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class DefaultBackgroundExecutorTest {
 
   @Test
   public void submit_callable() throws Exception {
 
-    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 2, "test");
+    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 2, "test", null);
 
     final Future<String> future0 = es.submit(() -> "Hello");
     final Future<String> future1 = es.submit(() -> "There");
@@ -39,7 +43,7 @@ public class DefaultBackgroundExecutorTest {
   public void shutdown_slowCallable_expect_interrupted() throws Exception {
 
     int shutdownWaitSecs = 1;
-    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, shutdownWaitSecs, "test");
+    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, shutdownWaitSecs, "test", null);
 
     final Future<String> future2 = es.submit(() -> {
       try {
@@ -61,7 +65,7 @@ public class DefaultBackgroundExecutorTest {
   @Disabled("test takes long time")
   public void shutdown_when_running_expect_waitAndNiceShutdown() {
 
-    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 20, "test");
+    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 20, "test", null);
 
     es.execute(new RunFor(3000, "a"));
     es.execute(new RunFor(3000, "b"));
@@ -74,7 +78,7 @@ public class DefaultBackgroundExecutorTest {
   @Disabled("test takes long time")
   public void shutdown_when_rougeRunnable_expect_InterruptedException() {
 
-    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test");
+    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test", null);
 
     es.execute(new RunFor(300000, "a"));
     es.execute(new RunFor(3000, "b"));
@@ -85,12 +89,12 @@ public class DefaultBackgroundExecutorTest {
 
   @Test
   public void wrapWithNoMDC() {
-    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test");
+    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test", null);
     assertThat(MDC.getCopyOfContextMap()).isNull();
-    es.wrapMDC(() -> {
+    es.wrap(() -> {
       assertThat(MDC.getCopyOfContextMap()).isNull();
     });
-    es.wrapMDC(() -> {
+    es.wrap(() -> {
       assertThat(MDC.getCopyOfContextMap()).isNull();
       return "Callable";
     });
@@ -98,25 +102,44 @@ public class DefaultBackgroundExecutorTest {
   }
 
   @Test
-  public void wrapWithMDC_expect_() {
-    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test");
+  public void wrapWithMDC_expect_() throws Exception {
+    DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test", new MdcBackgroundExecutorWrapper());
+    // MDC has a copyOnThread map. So we must pass different values to check if the test will work
     MDC.clear();
+    es.submit(()->{
+      assertThat(MDC.get("hello")).isNull();
+    }).get();
+    
     MDC.put("hello", "there");
-    es.wrapMDC(() -> {
+    es.wrap(() -> {
       assertThat(MDC.get("hello")).isEqualTo("there");
-    });
-    es.wrapMDC(() -> {
+    }).run(); // will clear the MDC. But this should be OK
+    
+    MDC.put("hello", "there");
+    es.wrap(() -> {
       assertThat(MDC.get("hello")).isEqualTo("there");
       return "Callable";
-    });
+    }).call(); // will clear the MDC. But this should be OK
+    
+    MDC.put("hello", "there");
+
+    CountDownLatch latch = new CountDownLatch(1);
     es.execute(() -> {
+      // the assertion is executed async, so it will only logged on console
       assertThat(MDC.get("hello")).isEqualTo("there");
+      latch.countDown();
     });
+    assertTrue(latch.await(5, TimeUnit.SECONDS));
+
     es.submit(() -> {
       assertThat(MDC.get("hello")).isEqualTo("there");
       return "Callable";
-    });
+    }).get();
     MDC.clear();
+    
+    es.execute(()->{
+      assertThat(MDC.get("hello")).isNull();
+    });
     es.shutdown();
   }
 

ebean-test/src/test/java/org/tests/cache/TestBeanCacheAsync.java

@@ -0,0 +1,113 @@
+package org.tests.cache;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import java.util.concurrent.Callable;
+
+import org.junit.jupiter.api.Test;
+import org.tests.model.basic.OCachedBean;
+
+import io.ebean.BaseTestCase;
+import io.ebean.DB;
+import io.ebean.Database;
+import io.ebean.DatabaseFactory;
+import io.ebean.config.CurrentTenantProvider;
+import io.ebean.config.DatabaseConfig;
+import io.ebean.config.MdcBackgroundExecutorWrapper;
+import io.ebeaninternal.server.cache.DefaultServerCachePlugin;
+
+/**
+ * Test class testing async/background cache updates in a multi-tenant environment. 
+ */
+public class TestBeanCacheAsync extends BaseTestCase {
+
+  private final ThreadLocal<String> tenantId = new ThreadLocal<>();
+
+  class ThreadLocalTenantProvider implements CurrentTenantProvider {
+
+    @Override
+    public Object currentId() {
+      return tenantId.get();
+    }
+  }
+  /**
+   * Copy tenant info to the background thread.
+   */
+  class TenantCopyBackgroundExecutorWrapper extends MdcBackgroundExecutorWrapper {
+    @Override
+    public <T> Callable<T> wrap(Callable<T> task) {
+      String tenant = tenantId.get();
+      if (tenant == null) {
+        return super.wrap(task);
+      } else {
+        return () -> {
+          tenantId.set(tenant);
+          try {
+            return super.wrap(task).call();
+          } finally {
+            tenantId.remove();
+          }
+        };
+      }
+    }
+    
+    @Override
+    public Runnable wrap(Runnable task) {
+      String tenant = tenantId.get();
+      if (tenant == null) {
+        return super.wrap(task);
+      } else {
+        return () -> {
+          tenantId.set(tenant);
+          try {
+            super.wrap(task).run();
+          } finally {
+            tenantId.remove();
+          }
+        };
+      }
+    }
+  }
+  
+  @Test
+  public void findById_with_tenant() throws InterruptedException {
+    DatabaseConfig config = new DatabaseConfig();
+    config.setName(DB.getDefault().name());
+    config.loadFromProperties();
+    config.setDataSource(DB.getDefault().dataSource());
+    config.setReadOnlyDataSource(DB.getDefault().readOnlyDataSource());
+    config.setDdlExtra(false);
+    config.setDdlGenerate(false);
+    config.setDdlRun(false);
+    config.setDefaultServer(false);
+    config.setRegister(false);
+    config.setServerCachePlugin(new DefaultServerCachePlugin()); // disables foreground local caching (as it is done in Hz/Ignite)
+    config.setCurrentTenantProvider(new ThreadLocalTenantProvider());
+    config.setBackgroundExecutorWrapper(new TenantCopyBackgroundExecutorWrapper());
+    tenantId.set("4711");
+    
+    Database db = DatabaseFactory.create(config);
+    try {
+      OCachedBean bean = new OCachedBean();
+      bean.setName("findById");
+      db.save(bean);
+
+      OCachedBean bean0 = db.find(OCachedBean.class, bean.getId());
+      assertNotNull(bean0);
+      assertThat(bean0.getName()).isEqualTo("findById");
+      bean0.setName("findById2");
+      db.save(bean0);
+
+      Thread.sleep(100); // TODO: can we block finds on that ID if a pending cache update is present?
+      
+      bean0 = db.find(OCachedBean.class, bean.getId());
+      assertNotNull(bean0);
+      assertThat(bean0.getName()).isEqualTo("findById2");
+
+    } finally {
+      db.shutdown();
+    }
+  }
+
+}