STAR-1246 Add unregister of compaction progress listener (apache#425)

(cherry picked from commit 7bacbfc)
(cherry picked from commit 4f4a59e)

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

@@ -62,6 +62,11 @@ public void registerListener(CompactionProgressListener listener)
         listeners.add(listener);
     }
 
+    public void unregisterListener(CompactionProgressListener listener)
+    {
+        listeners.remove(listener);
+    }
+
     /**
      * @return all the table operations currently in progress. This is mostly compactions but it can include other
      *         operations too, basically any operation that calls {@link this#onOperationStart(TableOperation).}

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

@@ -92,6 +92,8 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -681,8 +683,18 @@ public boolean isGlobal()
 
         Assert.assertEquals(totalByteScanned, operation.getProgress().totalByteScanned());
         try (NonThrowingCloseable cls = CompactionManager.instance.active.onOperationStart(operation))
-        {}
-        verify(listener).onStarted(progress);
-        verify(listener).onCompleted(progress);
+        {
+            verify(listener).onStarted(progress);
+            verify(listener, never()).onCompleted(progress);
+        }
+        verify(listener, times(1)).onStarted(progress);
+        verify(listener, times(1)).onCompleted(progress);
+
+        try (NonThrowingCloseable cls = CompactionManager.instance.active.onOperationStart(operation))
+        {
+            CompactionManager.instance.active.unregisterListener(listener);
+        }
+        verify(listener, times(2)).onStarted(progress);
+        verify(listener, times(1)).onCompleted(progress);
     }
 }