CNDB-14722: handle OOM error for compaction task (#1856)

### What is the issue

CNDB-14722: OOM error is not handled by error handler

https://github.com/riptano/cndb/pull/14726/files

### What does this PR fix and why was it fixed

Handle OOM error for compaction tasks

Author: jasonstack

src/java/org/apache/cassandra/db/compaction/BackgroundCompactionRunner.java

@@ -487,6 +487,11 @@ else if (Throwables.isCausedBy(t, IOException.class) && t.toString().contains(NO
             t = t instanceof FSError ? t : new FSWriteError(t);
             JVMStabilityInspector.inspectThrowable(t);
         }
+        else if (Throwables.isCausedBy(t, OutOfMemoryError.class))
+        {
+            logger.error("Encountered out of memory error on {}", cfs, t);
+            JVMStabilityInspector.inspectThrowable(t);
+        }
         else if (t instanceof CompactionInterruptedException)
         {
             logger.warn(String.format("Aborting background compaction of %s due to interruption", cfs), Throwables.unwrapped(t));

test/unit/org/apache/cassandra/db/compaction/BackgroundCompactionRunnerTest.java

@@ -28,6 +28,10 @@
 import com.google.common.collect.ImmutableSet;
 import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
 import org.apache.cassandra.concurrent.WrappedExecutorPlus;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.utils.JVMKiller;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.KillerForTests;
 import org.apache.cassandra.utils.concurrent.Promise;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -515,6 +519,25 @@ public void handleTaskFailure() throws Exception
         verifyCFSWasMarkedForCompaction();
     }
 
+    @Test
+    public void handleOOMError()
+    {
+        JVMKiller originalKiller = JVMStabilityInspector.replaceKiller(new KillerForTests());
+        Config.DiskFailurePolicy originalPolicy = DatabaseDescriptor.getDiskFailurePolicy();
+        try
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(Config.DiskFailurePolicy.die);
+
+            OutOfMemoryError oomError = new OutOfMemoryError("oom");
+            assertThatThrownBy(() -> BackgroundCompactionRunner.handleCompactionError(oomError, cfs))
+                    .isInstanceOf(OutOfMemoryError.class);
+        }
+        finally
+        {
+            DatabaseDescriptor.setDiskFailurePolicy(originalPolicy);
+            JVMStabilityInspector.replaceKiller(originalKiller);
+        }
+    }
 
     private void verifyTaskScheduled(Executor executor)
     {