Added additional parameter to JVM shutdown to allow for logs to be properly shutdown

patch by Alan Wang; reviewed by Abe Ratnofsky, Caleb Rackliffe for CASSANDRA-20978

Author: Alan Wang

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Added additional parameter to JVM shutdown to allow for logs to be properly shutdown (CASSANDRA-20978)
  * Improve isGossipOnlyMember and location lookup performance (CASSANDRA-21039)
  * Refactor the way we check if a transformation is allowed to be committed during upgrades (CASSANDRA-21043)
  * Improve debug around paused and disabled compaction (CASSANDRA-20131,CASSANDRA-19728)

src/java/org/apache/cassandra/service/DefaultDiskErrorsHandler.java

@@ -129,7 +129,7 @@ public void handleStartupFSError(Throwable t)
                 logger.error("Exiting forcefully due to file system exception on startup, disk failure policy \"{}\"",
                              DatabaseDescriptor.getDiskFailurePolicy(),
                              t);
-                JVMStabilityInspector.killCurrentJVM(t, true);
+                JVMStabilityInspector.killCurrentJVM(t, true, true);
                 break;
             default:
                 break;
@@ -174,10 +174,10 @@ public void inspectCommitLogError(Throwable t)
         if (!StorageService.instance.isDaemonSetupCompleted())
         {
             logger.error("Exiting due to error while processing commit log during initialization.", t);
-            JVMStabilityInspector.killCurrentJVM(t, true);
+            JVMStabilityInspector.killCurrentJVM(t, true, true);
         }
         else if (DatabaseDescriptor.getCommitFailurePolicy() == Config.CommitFailurePolicy.die)
-            JVMStabilityInspector.killCurrentJVM(t, false);
+            JVMStabilityInspector.killCurrentJVM(t, false, true);
     }
 
     private boolean shouldMaybeRemoveData(Throwable error)

src/java/org/apache/cassandra/utils/JVMStabilityInspector.java

@@ -47,6 +47,7 @@
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
+import org.apache.cassandra.utils.logging.LoggingSupportFactory;
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.PRINT_HEAP_HISTOGRAM_ON_OUT_OF_MEMORY_ERROR;
 
@@ -226,16 +227,21 @@ public static void killCurrentJVM(Throwable t, boolean quiet)
         killer.killCurrentJVM(t, quiet);
     }
 
+    public static void killCurrentJVM(Throwable t, boolean quiet, boolean callShutDownOnLogger)
+    {
+        killer.killCurrentJVM(t, quiet, callShutDownOnLogger);
+    }
+
     public static void userFunctionTimeout(Throwable t)
     {
         switch (DatabaseDescriptor.getUserFunctionTimeoutPolicy())
         {
             case die:
                 // policy to give 250ms grace time to
-                ScheduledExecutors.nonPeriodicTasks.schedule(() -> killer.killCurrentJVM(t), 250, TimeUnit.MILLISECONDS);
+                ScheduledExecutors.nonPeriodicTasks.schedule(() -> killer.killCurrentJVM(t, false, true), 250, TimeUnit.MILLISECONDS);
                 break;
             case die_immediate:
-                killer.killCurrentJVM(t);
+                killer.killCurrentJVM(t, false, true);
                 break;
             case ignore:
                 logger.error(t.getMessage());
@@ -268,6 +274,11 @@ public void killCurrentJVM(Throwable t)
         }
 
         public void killCurrentJVM(Throwable t, boolean quiet)
+        {
+            killCurrentJVM(t, quiet, false);
+        }
+
+        public void killCurrentJVM(Throwable t, boolean quiet, boolean callShutDownOnLogger)
         {
             if (!quiet)
             {
@@ -279,6 +290,8 @@ public void killCurrentJVM(Throwable t, boolean quiet)
 
             if (doExit && killing.compareAndSet(false, true))
             {
+                if (callShutDownOnLogger)
+                    LoggingSupportFactory.getLoggingSupport().onShutdown();
                 StorageService.instance.removeShutdownHook();
                 System.exit(100);
             }

test/unit/org/apache/cassandra/cql3/OutOfSpaceTest.java

@@ -79,6 +79,7 @@ public void testFlushUnwriteableDie() throws Throwable
             flushAndExpectError();
             Assert.assertTrue(killerForTests.wasKilled());
             Assert.assertFalse(killerForTests.wasKilledQuietly()); //only killed quietly on startup failure
+            Assert.assertFalse(killerForTests.calledShutDownOnLogger());
         }
         finally
         {

test/unit/org/apache/cassandra/db/commitlog/CommitLogFailurePolicyTest.java

@@ -83,6 +83,7 @@ public void testCommitFailurePolicy_die()
             CommitLog.handleCommitError("Testing die policy", new Throwable());
             Assert.assertTrue(killerForTests.wasKilled());
             Assert.assertFalse(killerForTests.wasKilledQuietly()); //only killed quietly on startup failure
+            Assert.assertTrue(killerForTests.calledShutDownOnLogger());
         }
         finally
         {
@@ -108,6 +109,7 @@ public void testCommitFailurePolicy_ignore_beforeStartup()
             //even though policy is ignore, JVM must die because Daemon has not finished initializing
             Assert.assertTrue(killerForTests.wasKilled());
             Assert.assertTrue(killerForTests.wasKilledQuietly()); //killed quietly due to startup failure
+            Assert.assertTrue(killerForTests.calledShutDownOnLogger());
         }
         finally
         {

test/unit/org/apache/cassandra/net/ResourceLimitsTest.java

@@ -173,6 +173,7 @@ public void negativeConcurrentUsingValueKillsJVMTest()
             }
             Assert.assertTrue(killerForTests.wasKilled());
             Assert.assertFalse(killerForTests.wasKilledQuietly()); //only killed quietly on startup failure
+            Assert.assertFalse(killerForTests.calledShutDownOnLogger());
         }
         finally
         {

test/unit/org/apache/cassandra/utils/JVMStabilityInspectorTest.java

@@ -118,6 +118,48 @@ public void testKill() throws Exception
         }
     }
 
+    @Test
+    public void testCallShutDownOnLoggerOnSpecificErrors() throws Exception
+    {
+        KillerForTests killerForTests = new KillerForTests();
+        JVMStabilityInspector.Killer originalKiller = JVMStabilityInspector.replaceKiller(killerForTests);
+
+        Config.DiskFailurePolicy oldPolicy = DatabaseDescriptor.getDiskFailurePolicy();
+        Config.UserFunctionTimeoutPolicy oldUserFunctionTimeoutPolicy = DatabaseDescriptor.getUserFunctionTimeoutPolicy();
+        DiskErrorsHandlerService.configure();
+        try
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(Config.DiskFailurePolicy.die);
+            killerForTests.reset();
+            JVMStabilityInspector.inspectThrowable(new FSWriteError(new IOException(), "blah"));
+            assertTrue(killerForTests.wasKilled());
+            assertTrue(killerForTests.calledShutDownOnLogger());
+
+            killerForTests.reset();
+            JVMStabilityInspector.inspectCommitLogThrowable(new Throwable());
+            assertTrue(killerForTests.wasKilled());
+            assertTrue(killerForTests.calledShutDownOnLogger());
+
+            DatabaseDescriptor.setUserFunctionTimeoutPolicy(Config.UserFunctionTimeoutPolicy.die_immediate);
+            killerForTests.reset();
+            JVMStabilityInspector.userFunctionTimeout(new Throwable());
+            assertTrue(killerForTests.wasKilled());
+            assertTrue(killerForTests.calledShutDownOnLogger());
+        }
+        catch (Exception | Error e)
+        {
+            throw new AssertionError("Failure when daemonSetupCompleted=false", e);
+        }
+        finally
+        {
+            JVMStabilityInspector.replaceKiller(originalKiller);
+            DatabaseDescriptor.setDiskFailurePolicy(oldPolicy);
+            DatabaseDescriptor.setUserFunctionTimeoutPolicy(oldUserFunctionTimeoutPolicy);
+            StorageService.instance.registerDaemon(null);
+            DiskErrorsHandlerService.set(DiskErrorsHandler.NoOpDiskErrorHandler.NO_OP);
+        }
+    }
+
     @Test
     public void testOutOfMemoryHandling()
     {

test/unit/org/apache/cassandra/utils/KillerForTests.java

@@ -27,6 +27,7 @@ public class KillerForTests extends JVMStabilityInspector.Killer
 {
     private boolean killed = false;
     private boolean quiet = false;
+    private boolean callShutDownOnLogger = false;
     private final boolean expected;
 
     public KillerForTests()
@@ -41,6 +42,12 @@ public KillerForTests(boolean expectFailure)
 
     @Override
     public void killCurrentJVM(Throwable t, boolean quiet)
+    {
+        killCurrentJVM(t, quiet, false);
+    }
+
+    @Override
+    public void killCurrentJVM(Throwable t, boolean quiet, boolean callShutDownOnLogger)
     {
         if (!expected)
             Assert.fail("Saw JVM Kill but did not expect it.");
@@ -52,6 +59,7 @@ public void killCurrentJVM(Throwable t, boolean quiet)
         }
         this.killed = true;
         this.quiet = quiet;
+        this.callShutDownOnLogger = callShutDownOnLogger;
     }
 
     public boolean wasKilled()
@@ -64,8 +72,15 @@ public boolean wasKilledQuietly()
         return quiet;
     }
 
+    public boolean calledShutDownOnLogger()
+    {
+        return callShutDownOnLogger;
+    }
+
     public void reset()
     {
         killed = false;
+        quiet = false;
+        callShutDownOnLogger = false;
     }
 }