WIP - Improved HA stability

client/src/main/java/com/orientechnologies/orient/core/db/OrientDBRemote.java

@@ -445,6 +445,11 @@ public void loadAllDatabases() {
     // In remote does nothing
   }
 
+  @Override
+  public ODatabaseDocumentInternal openInternal(String iDbUrl, String user) {
+    throw new UnsupportedOperationException("Open for internal use is not supported in remote");
+  }
+
   @Override
   public ODatabaseDocumentInternal openNoAuthenticate(String iDbUrl, String user) {
     throw new UnsupportedOperationException(

core/src/main/java/com/orientechnologies/common/thread/OScheduledThreadPoolExecutorWithLogging.java

@@ -1,18 +1,14 @@
 package com.orientechnologies.common.thread;
 
 import com.orientechnologies.common.log.OLogManager;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionHandler;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.*;
 
 /**
  * The same as thread {@link ScheduledThreadPoolExecutor} but also logs all exceptions happened
  * inside of the tasks which caused tasks to stop.
  */
-public class OScheduledThreadPoolExecutorWithLogging extends ScheduledThreadPoolExecutor {
+public class OScheduledThreadPoolExecutorWithLogging extends ScheduledThreadPoolExecutor
+    implements TracingScheduledExecutorService {
   public OScheduledThreadPoolExecutorWithLogging(int corePoolSize) {
     super(corePoolSize);
   }
@@ -58,4 +54,168 @@ protected void afterExecute(Runnable r, Throwable t) {
       OLogManager.instance().errorNoDb(this, "Exception in thread '%s'", t, thread.getName());
     }
   }
+
+  @Override
+  public <T> Future<T> submit(String taskName, Callable<T> task) {
+    final OTracedExecutionException trace = OTracedExecutionException.prepareTrace(taskName, task);
+    return super.submit(
+        () -> {
+          try {
+            return task.call();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, task);
+          }
+        });
+  }
+
+  @Override
+  public Future<?> submit(String taskName, Runnable task) {
+    final OTracedExecutionException trace = OTracedExecutionException.prepareTrace(taskName, task);
+    return super.submit(
+        () -> {
+          try {
+            task.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, task);
+          }
+        });
+  }
+
+  @Override
+  public <T> Future<T> submit(String taskName, Runnable task, T result) {
+    final OTracedExecutionException trace = OTracedExecutionException.prepareTrace(taskName, task);
+    return super.submit(
+        () -> {
+          try {
+            task.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, task);
+          }
+        },
+        result);
+  }
+
+  @Override
+  public void execute(String taskName, Runnable command) {
+    final OTracedExecutionException trace =
+        OTracedExecutionException.prepareTrace(taskName, command);
+    super.execute(
+        () -> {
+          try {
+            command.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, command);
+          }
+        });
+  }
+
+  @Override
+  public Future<?> submit(Runnable task) {
+    return submit((String) null, task);
+  }
+
+  @Override
+  public <T> Future<T> submit(Callable<T> task) {
+    return submit((String) null, task);
+  }
+
+  @Override
+  public <T> Future<T> submit(Runnable task, T result) {
+    return submit(null, task, result);
+  }
+
+  @Override
+  public void execute(Runnable command) {
+    execute(null, command);
+  }
+
+  @Override
+  public ScheduledFuture<?> schedule(String taskName, Runnable command, long delay, TimeUnit unit) {
+    final OTracedExecutionException trace =
+        OTracedExecutionException.prepareTrace(taskName, command);
+    return super.schedule(
+        () -> {
+          try {
+            command.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, command);
+          }
+        },
+        delay,
+        unit);
+  }
+
+  @Override
+  public <V> ScheduledFuture<V> schedule(
+      String taskName, Callable<V> task, long delay, TimeUnit unit) {
+    final OTracedExecutionException trace = OTracedExecutionException.prepareTrace(taskName, task);
+    return super.schedule(
+        () -> {
+          try {
+            return task.call();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, task);
+          }
+        },
+        delay,
+        unit);
+  }
+
+  @Override
+  public ScheduledFuture<?> scheduleAtFixedRate(
+      String taskName, Runnable command, long initialDelay, long period, TimeUnit unit) {
+    final OTracedExecutionException trace =
+        OTracedExecutionException.prepareTrace(taskName, command);
+    return super.scheduleAtFixedRate(
+        () -> {
+          try {
+            command.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, command);
+          }
+        },
+        initialDelay,
+        period,
+        unit);
+  }
+
+  @Override
+  public ScheduledFuture<?> scheduleWithFixedDelay(
+      String taskName, Runnable command, long initialDelay, long delay, TimeUnit unit) {
+    final OTracedExecutionException trace =
+        OTracedExecutionException.prepareTrace(taskName, command);
+    return super.scheduleWithFixedDelay(
+        () -> {
+          try {
+            command.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, command);
+          }
+        },
+        initialDelay,
+        delay,
+        unit);
+  }
+
+  @Override
+  public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+    return schedule(null, command, delay, unit);
+  }
+
+  @Override
+  public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+    return schedule(null, callable, delay, unit);
+  }
+
+  @Override
+  public ScheduledFuture<?> scheduleAtFixedRate(
+      Runnable command, long initialDelay, long period, TimeUnit unit) {
+    return scheduleAtFixedRate(null, command, initialDelay, period, unit);
+  }
+
+  @Override
+  public ScheduledFuture<?> scheduleWithFixedDelay(
+      Runnable command, long initialDelay, long delay, TimeUnit unit) {
+    return scheduleWithFixedDelay(null, command, initialDelay, delay, unit);
+  }
 }

core/src/main/java/com/orientechnologies/common/thread/OThreadPoolExecutorWithLogging.java

@@ -2,6 +2,7 @@
 
 import com.orientechnologies.common.log.OLogManager;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -14,7 +15,8 @@
  * The same as thread {@link ThreadPoolExecutor} but also logs all exceptions happened inside of the
  * tasks which caused tasks to stop.
  */
-public class OThreadPoolExecutorWithLogging extends ThreadPoolExecutor {
+public class OThreadPoolExecutorWithLogging extends ThreadPoolExecutor
+    implements TracingExecutorService {
   public OThreadPoolExecutorWithLogging(
       int corePoolSize,
       int maximumPoolSize,
@@ -79,4 +81,78 @@ protected void afterExecute(Runnable r, Throwable t) {
       OLogManager.instance().errorNoDb(this, "Exception in thread '%s'", t, thread.getName());
     }
   }
+
+  @Override
+  public <T> Future<T> submit(String taskName, Callable<T> task) {
+    final OTracedExecutionException trace = OTracedExecutionException.prepareTrace(taskName, task);
+    return super.submit(
+        () -> {
+          try {
+            return task.call();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, task);
+          }
+        });
+  }
+
+  @Override
+  public Future<?> submit(String taskName, Runnable task) {
+    final OTracedExecutionException trace = OTracedExecutionException.prepareTrace(taskName, task);
+    return super.submit(
+        () -> {
+          try {
+            task.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, task);
+          }
+        });
+  }
+
+  @Override
+  public <T> Future<T> submit(String taskName, Runnable task, T result) {
+    final OTracedExecutionException trace = OTracedExecutionException.prepareTrace(taskName, task);
+    return super.submit(
+        () -> {
+          try {
+            task.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, task);
+          }
+        },
+        result);
+  }
+
+  @Override
+  public void execute(String taskName, Runnable command) {
+    final OTracedExecutionException trace =
+        OTracedExecutionException.prepareTrace(taskName, command);
+    super.execute(
+        () -> {
+          try {
+            command.run();
+          } catch (Exception e) {
+            throw OTracedExecutionException.trace(trace, e, taskName, command);
+          }
+        });
+  }
+
+  @Override
+  public Future<?> submit(Runnable task) {
+    return submit((String) null, task);
+  }
+
+  @Override
+  public <T> Future<T> submit(Callable<T> task) {
+    return submit((String) null, task);
+  }
+
+  @Override
+  public <T> Future<T> submit(Runnable task, T result) {
+    return submit(null, task, result);
+  }
+
+  @Override
+  public void execute(Runnable command) {
+    execute(null, command);
+  }
 }