Merge pull request #924: [proxima-direct-transaction-manager] #337 en…

…sure transaction server terminates on errors
datadrivencz · Jul 8, 2024 · 5426299 · 5426299
2 parents c9947c7 + 503d6c9
commit 5426299
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 8 deletions.
diff --git a/...ager/src/main/java/cz/o2/proxima/direct/transaction/manager/TransactionManagerServer.java b/...ager/src/main/java/cz/o2/proxima/direct/transaction/manager/TransactionManagerServer.java
@@ -27,6 +27,7 @@
 import cz.o2.proxima.typesafe.config.Config;
 import cz.o2.proxima.typesafe.config.ConfigFactory;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -61,13 +62,15 @@ public static void main(String[] args) {
     }
     TransactionManagerServer server = TransactionManagerServer.of(config);
     try {
-      Runtime.getRuntime().addShutdownHook(new Thread(server::stop));
+      Runtime.getRuntime().addShutdownHook(new Thread(() -> server.stop(true)));
       server.run();
       while (!Thread.currentThread().isInterrupted() && !server.isStopped()) {
         ExceptionUtils.ignoringInterrupted(() -> TimeUnit.SECONDS.sleep(10));
       }
-    } finally {
-      server.stop();
+    } catch (Throwable err) {
+      log.error(
+          "Exception caught while running {}", TransactionManagerServer.class.getSimpleName(), err);
+      server.stop(false);
     }
   }
 
@@ -105,7 +108,7 @@ private TransactionLogObserverFactory getObserverFactory(Config conf) {
     return new TransactionLogObserverFactory.WithOnErrorHandler(
         error -> {
           log.error("Error processing transactions. Bailing out for safety.", error);
-          asyncTerminate(this::stop, () -> System.exit(1));
+          asyncTerminate(() -> stop(false), () -> System.exit(1));
         });
   }
 
@@ -139,11 +142,18 @@ private TransactionLogObserver newTransactionLogObserver() {
     return observerFactory.create(direct, metrics);
   }
 
-  public void stop() {
+  public void stop(boolean graceful) {
     if (closed.compareAndSet(false, true)) {
       log.info("{} shutting down.", getClass().getSimpleName());
-      manager.close();
-      direct.close();
+      CompletableFuture<Void> shutdownFuture =
+          CompletableFuture.runAsync(
+              () -> {
+                manager.close();
+                direct.close();
+              });
+      ExceptionUtils.ignoringInterrupted(() -> shutdownFuture.get(1, TimeUnit.SECONDS));
+      log.info("{} halting now.", getClass().getSimpleName());
+      System.exit(graceful ? 0 : 1);
     }
   }
 

diff --git a/.../src/test/java/cz/o2/proxima/direct/transaction/manager/TransactionManagerServerTest.java b/.../src/test/java/cz/o2/proxima/direct/transaction/manager/TransactionManagerServerTest.java
@@ -51,7 +51,7 @@ void validateModeSupported(Repository repo) {
   public void testServerRunTearDown() {
     server.run();
     assertFalse(server.isStopped());
-    server.stop();
+    server.stop(true);
     assertTrue(server.isStopped());
   }