CNDB-12774: change CompactionObserver#onCompleted to include failure exception (#1573)

https://jenkins-stargazer.aws.dsinternal.org/view/cc-builds/job/ds-cassandra-publish-j11/1293/

Author: jasonstack

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

@@ -119,7 +119,7 @@ public void onInProgress(CompactionProgress progress)
     }
 
     @Override
-    public void onCompleted(TimeUUID id, boolean isSuccess)
+    public void onCompleted(TimeUUID id, Throwable err)
     {
         backgroundCompactions.onCompleted(this, id);
     }

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

@@ -160,9 +160,9 @@ public Throwable rejected(Throwable t)
 
     protected Throwable cleanup(Throwable err)
     {
-        final boolean isSuccess = err == null;
+        final Throwable originalError = err;
         for (CompactionObserver compObserver : compObservers)
-            err = Throwables.perform(err, () -> compObserver.onCompleted(transaction.opId(), isSuccess));
+            err = Throwables.perform(err, () -> compObserver.onCompleted(transaction.opId(), originalError));
 
         return Throwables.perform(err, () -> transaction.close());
     }

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

@@ -20,6 +20,7 @@
 
 import org.apache.cassandra.utils.TimeUUID;
 
+import javax.annotation.Nullable;
 
 /**
  * An observer of a compaction operation. It is notified when a compaction operation is started.
@@ -36,7 +37,7 @@ public interface CompactionObserver
         public void onInProgress(CompactionProgress progress) { }
 
         @Override
-        public void onCompleted(TimeUUID id, boolean isSuccess) { }
+        public void onCompleted(TimeUUID id, @Nullable Throwable error) { }
     };
 
     /**
@@ -50,7 +51,7 @@ public void onCompleted(TimeUUID id, boolean isSuccess) { }
      * Indicates that a compaction with the given id has completed.
      * <p/>
      * @param id  the id of the compaction
-     * @param isSuccess true if compaction finished without any exceptions
+     * @param error error if compaction failed with any exceptions; or null if completed successfully
      */
-    void onCompleted(TimeUUID id, boolean isSuccess);
+    void onCompleted(TimeUUID id, @Nullable Throwable error);
 }

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

@@ -1286,7 +1286,7 @@ public void onInProgress(CompactionProgress progress)
     }
 
     @Override
-    public void onCompleted(TimeUUID id, boolean isSuccess)
+    public void onCompleted(TimeUUID id, Throwable err)
     {
 
     }

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

@@ -18,7 +18,6 @@
 
 package org.apache.cassandra.db.compaction;
 
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -36,7 +35,7 @@
 public class SharedCompactionObserver implements CompactionObserver
 {
     private final AtomicInteger toReportOnComplete = new AtomicInteger(0);
-    private final AtomicBoolean onCompleteIsSuccess = new AtomicBoolean(true);
+    private final AtomicReference<Throwable> onCompleteException = new AtomicReference(null);
     private final AtomicReference<CompactionProgress> inProgressReported = new AtomicReference<>(null);
     private final CompactionObserver observer;
 
@@ -62,14 +61,14 @@ public void onInProgress(CompactionProgress progress)
     }
 
     @Override
-    public void onCompleted(TimeUUID id, boolean isSuccess)
+    public void onCompleted(TimeUUID id, Throwable err)
     {
-        onCompleteIsSuccess.compareAndSet(true, isSuccess); // acts like AND
+        onCompleteException.compareAndSet(null, err); // acts like AND
         final int remainingToComplete = toReportOnComplete.decrementAndGet();
         assert inProgressReported.get() != null : "onCompleted called before onInProgress";
         assert remainingToComplete >= 0 : "onCompleted called without corresponding registerExpectedSubtask";
         // The individual operation ID given here may be different from the shared ID. Pass on the shared one.
         if (remainingToComplete == 0)
-            observer.onCompleted(inProgressReported.get().operationId(), onCompleteIsSuccess.get());
+            observer.onCompleted(inProgressReported.get().operationId(), onCompleteException.get());
     }
 }

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

@@ -397,9 +397,9 @@ public void onInProgress(CompactionProgress progress)
     }
 
     @Override
-    public void onCompleted(TimeUUID id, boolean isSuccess)
+    public void onCompleted(TimeUUID id, Throwable err)
     {
-        strategy.onCompleted(id, isSuccess);
+        strategy.onCompleted(id, err);
     }
 
     @Override

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

@@ -30,7 +30,6 @@
 import java.util.function.BiPredicate;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;

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

@@ -1382,7 +1382,7 @@ private void compactData()
                     //Thread.sleep(5);
 
                     // then remove the old sstables
-                    strategy.onCompleted(id, true);
+                    strategy.onCompleted(id, null);
                     counters.numCompactions.incrementAndGet();
                     counters.numCompactionsPending.decrementAndGet();
                     counters.numCompactedSSTables.addAndGet(candidates.size());

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

@@ -663,7 +663,7 @@ private void testCompactionStatistics(Set<SSTableReader> sstables,
         {
             Set<SSTableReader> compSSTables = pair.left;
             long totSSTablesLen = totUncompressedLength(compSSTables);
-            strategy.onCompleted(pair.right, true);
+            strategy.onCompleted(pair.right, null);
 
             numCompactions--;
             numCompactionsInProgress--;

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

@@ -340,7 +340,7 @@ public void testCompactionReporting()
 
         verify(operationObserver, times(1)).onOperationStart(tableOpCaptor.capture());
         verify(compObserver, times(1)).onInProgress(compactionCaptor.capture());
-        verify(compObserver, times(1)).onCompleted(eq(txn.opId()), eq(true));
+        verify(compObserver, times(1)).onCompleted(eq(txn.opId()), eq(null));
     }
 
     @Test
@@ -354,7 +354,7 @@ public void testFailCompactionTask()
         taskMock.addObserver(compObserver);
         Mockito.doThrow(new RuntimeException("Test throw")).when(taskMock).executeInternal();
         Assert.assertThrows(RuntimeException.class, () ->  taskMock.execute());
-        Mockito.verify(compObserver, times(1)).onCompleted(any(TimeUUID.class), eq(false));
+        Mockito.verify(compObserver, times(1)).onCompleted(any(TimeUUID.class), any(Throwable.class));
     }
 
     private Set<SSTableReader> generateData(int numSSTables, int numKeys)