CASSANDRA-19576 Write compaction history asynchronously through a bounded executor

CHANGES.txt

@@ -1,4 +1,5 @@
 7.0
+ * Isolate asynchronous best-effort compaction history writes with a bounded queue and drain them before system-table shutdown (CASSANDRA-19576)
  * Allow CQLSSTableWriter to specify SSTable id generator to use (CASSANDRA-21012)
  * Reject LIKE patterns with a wildcard (%) anywhere other than the start or end (CASSANDRA-21068)
  * Support pluggable default role initialization (CASSANDRA-21546)

src/java/org/apache/cassandra/db/CompactionHistoryWriter.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.cassandra.db;
+
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.concurrent.ExecutorBuilder;
+import org.apache.cassandra.concurrent.ExecutorPlus;
+import org.apache.cassandra.utils.ExecutorUtils;
+import org.apache.cassandra.utils.NoSpamLogger;
+import org.apache.cassandra.utils.concurrent.Future;
+import org.apache.cassandra.utils.concurrent.ImmediateFuture;
+
+/** Isolates best-effort history writes from compactions and shared maintenance schedulers. */
+final class CompactionHistoryWriter
+{
+    private static final Logger logger = LoggerFactory.getLogger(CompactionHistoryWriter.class);
+    private final ExecutorPlus executor;
+
+    CompactionHistoryWriter(int queueLimit, ExecutorBuilder<? extends ExecutorPlus> builder)
+    {
+        executor = builder.withQueueLimit(queueLimit)
+                          .withRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy())
+                          .build();
+    }
+
+    Future<Void> submit(Runnable write)
+    {
+        try
+        {
+            return executor.submit(write, null);
+        }
+        catch (RejectedExecutionException e)
+        {
+            // Drop the newest record, never block the compaction caller or run a write on its thread.
+            NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1, TimeUnit.MINUTES,
+                             "Compaction history queue is full or shut down; dropping history update");
+            return ImmediateFuture.failure(e);
+        }
+    }
+
+    void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
+    {
+        // Drain accepted writes before the system memtables and commitlog are shut down. A timeout must
+        // fail drain rather than allow a still-running history mutation to race commitlog shutdown.
+        ExecutorUtils.shutdownAndWait(timeout, unit, executor);
+    }
+}

src/java/org/apache/cassandra/db/SystemKeyspace.java

@@ -36,6 +36,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
@@ -132,11 +133,13 @@
 import org.apache.cassandra.utils.TimeUUID;
 import org.apache.cassandra.utils.TriFunction;
 import org.apache.cassandra.utils.concurrent.Future;
+import org.apache.cassandra.utils.concurrent.ImmediateFuture;
 
 import static java.lang.String.format;
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
 import static java.util.concurrent.TimeUnit.MICROSECONDS;
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
 import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
 import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
 import static org.apache.cassandra.cql3.QueryProcessor.PREPARED_STATEMENT_CACHE_SIZE_BYTES;
@@ -713,7 +716,18 @@ public static void persistLocalMetadata()
                             DatabaseDescriptor.getStoragePort());
     }
 
-    public static void updateCompactionHistory(TimeUUID taskId,
+    // Like tracing/sampling, history is diagnostic: cap pending records rather than backpressure compactions.
+    private static final int COMPACTION_HISTORY_QUEUE_LIMIT = 1000;
+    @VisibleForTesting
+    static final CompactionHistoryWriter compactionHistoryWriter =
+        new CompactionHistoryWriter(COMPACTION_HISTORY_QUEUE_LIMIT, executorFactory().withJmxInternal().configureSequential("CompactionHistory"));
+
+    /**
+     * Enqueues a best-effort history write. Compaction completion does not imply history visibility.
+     * The returned future completes after insertion, or exceptionally on write failure or rejection
+     * (queue full or shutdown). Rejection drops history, never the compacted data.
+     */
+    public static Future<Void> updateCompactionHistory(TimeUUID taskId,
                                                String ksname,
                                                String cfname,
                                                long compactedAt,
@@ -724,17 +738,27 @@ public static void updateCompactionHistory(TimeUUID taskId,
     {
         // don't write anything when the history table itself is compacted, since that would in turn cause new compactions
         if (ksname.equals("system") && cfname.equals(COMPACTION_HISTORY))
-            return;
-        String req = "INSERT INTO system.%s (id, keyspace_name, columnfamily_name, compacted_at, bytes_in, bytes_out, rows_merged, compaction_properties) VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
-        executeInternal(format(req, COMPACTION_HISTORY),
-                        taskId,
-                        ksname,
-                        cfname,
-                        ByteBufferUtil.bytes(compactedAt),
-                        bytesIn,
-                        bytesOut,
-                        rowsMerged,
-                        compactionProperties);
+            return ImmediateFuture.success(null);
+
+        Map<Integer, Long> rows = ImmutableMap.copyOf(rowsMerged);
+        Map<String, String> properties = ImmutableMap.copyOf(compactionProperties);
+        return compactionHistoryWriter.submit(() -> {
+            String req = "INSERT INTO system.%s (id, keyspace_name, columnfamily_name, compacted_at, bytes_in, bytes_out, rows_merged, compaction_properties) VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
+            executeInternal(format(req, COMPACTION_HISTORY),
+                            taskId,
+                            ksname,
+                            cfname,
+                            ByteBufferUtil.bytes(compactedAt),
+                            bytesIn,
+                            bytesOut,
+                            rows,
+                            properties);
+        });
+    }
+
+    public static void shutdownCompactionHistoryAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
+    {
+        compactionHistoryWriter.shutdownAndWait(timeout, unit);
     }
 
     public static TabularData getCompactionHistory() throws OpenDataException

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

@@ -4018,6 +4018,7 @@ protected synchronized void drain(boolean isFinalShutdown) throws IOException, I
 
             // Interrupt ongoing compactions and shutdown CM to prevent further compactions.
             CompactionManager.instance.forceShutdown();
+            SystemKeyspace.shutdownCompactionHistoryAndWait(DRAIN_EXECUTOR_TIMEOUT_MS.getInt(), TimeUnit.MILLISECONDS);
             // Flush the system tables after all other tables are flushed, just in case flushing modifies any system state
             // like CASSANDRA-5151. Don't bother with progress tracking since system data is tiny.
             // Flush system tables after stopping compactions since they modify

test/distributed/org/apache/cassandra/distributed/impl/Instance.java

@@ -977,6 +977,9 @@ public Future<Void> shutdown(boolean runOnExitThreads, boolean shutdownMessaging
             error = parallelRun(error, executor, SnapshotManager.instance::close);
 
             CompactionManager.instance.forceShutdown();
+            // This lifecycle bypasses StorageService.drain. Stop history writes before their
+            // memtable/commitlog dependencies, including shutdowns that skip on-exit threads.
+            error = parallelRun(error, executor, () -> SystemKeyspace.shutdownCompactionHistoryAndWait(1L, MINUTES));
 
             error = parallelRun(error, executor,
                     () -> StorageService.instance.setRpcReady(false),

test/distributed/org/apache/cassandra/distributed/test/CompactionHistoryShutdownTest.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.cassandra.distributed.test;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
+import static org.junit.Assert.assertEquals;
+
+public class CompactionHistoryShutdownTest extends TestBaseImpl
+{
+    @Test
+    public void testHistoryWriterShutdownWithOnExitThreads() throws Exception
+    {
+        try (Cluster cluster = Cluster.build(1).start())
+        {
+            cluster.get(1).runOnInstance(() -> {
+                TimeUUID id = TimeUUID.Generator.nextTimeUUID();
+                FBUtilities.waitOnFuture(SystemKeyspace.updateCompactionHistory(id, "history_shutdown_ks", "history_shutdown_cf",
+                                                                               System.currentTimeMillis(), 1000, 500,
+                                                                               Collections.singletonMap(1, 100L),
+                                                                               Collections.singletonMap("strategy", "STCS")),
+                                         Duration.ofSeconds(10));
+                assertEquals(1, executeInternal("SELECT id FROM system.compaction_history WHERE id=?", id).size());
+            });
+
+            // Successful history insertion starts the persistent worker. Instance.shutdown must stop
+            // it independently of StorageService.drain and pass the harness's thread-leak assertion.
+            cluster.get(1).shutdown(true).get(2, TimeUnit.MINUTES);
+        }
+    }
+}

test/unit/org/apache/cassandra/db/CompactionHistoryWriterTest.java

@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.cassandra.db;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.utils.concurrent.Future;
+
+import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class CompactionHistoryWriterTest
+{
+    @BeforeClass
+    public static void setup()
+    {
+        SchemaLoader.prepareServer();
+    }
+
+    @Test(timeout = 30000)
+    public void testBoundedBacklogRejectsNewestAndPreservesFifo() throws Exception
+    {
+        CompactionHistoryWriter writer = new CompactionHistoryWriter(1, executorFactory().configureSequential("HistoryBacklogTest"));
+        CountDownLatch started = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        List<Integer> writes = new ArrayList<>();
+        try
+        {
+            Future<Void> first = writer.submit(() -> {
+                started.countDown();
+                await(release);
+                writes.add(1);
+            });
+            assertTrue(started.await(10, TimeUnit.SECONDS));
+            Future<Void> second = writer.submit(() -> writes.add(2));
+            Future<Void> rejected = writer.submit(() -> writes.add(3));
+            assertFailure(rejected, RejectedExecutionException.class);
+            assertFalse(first.isDone());
+            assertFalse(second.isDone());
+            release.countDown();
+            first.get(10, TimeUnit.SECONDS);
+            second.get(10, TimeUnit.SECONDS);
+            assertEquals(Arrays.asList(1, 2), writes);
+            writer.submit(() -> writes.add(4)).get(10, TimeUnit.SECONDS);
+            assertEquals(Arrays.asList(1, 2, 4), writes);
+        }
+        finally
+        {
+            release.countDown();
+            writer.shutdownAndWait(10, TimeUnit.SECONDS);
+        }
+    }
+
+    @Test(timeout = 30000)
+    public void testShutdownDrainsAcceptedWritesAndRejectsNewWrites() throws Exception
+    {
+        CompactionHistoryWriter writer = new CompactionHistoryWriter(1, executorFactory().configureSequential("HistoryShutdownTest"));
+        CountDownLatch started = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        List<Integer> writes = new ArrayList<>();
+        try
+        {
+            Future<Void> first = writer.submit(() -> {
+                started.countDown();
+                await(release);
+                writes.add(1);
+            });
+            assertTrue(started.await(10, TimeUnit.SECONDS));
+            Future<Void> second = writer.submit(() -> writes.add(2));
+            try
+            {
+                writer.shutdownAndWait(0, TimeUnit.NANOSECONDS);
+                fail("A blocked history mutation must prevent successful drain");
+            }
+            catch (TimeoutException expected)
+            {
+                assertFalse(first.isDone());
+                assertFalse(second.isDone());
+            }
+            assertFailure(writer.submit(() -> writes.add(3)), RejectedExecutionException.class);
+            release.countDown();
+            writer.shutdownAndWait(10, TimeUnit.SECONDS);
+            first.get(10, TimeUnit.SECONDS);
+            second.get(10, TimeUnit.SECONDS);
+            assertEquals(Arrays.asList(1, 2), writes);
+        }
+        finally
+        {
+            release.countDown();
+            writer.shutdownAndWait(10, TimeUnit.SECONDS);
+        }
+    }
+
+    private static Throwable assertFailure(Future<Void> future, Class<? extends Throwable> type) throws Exception
+    {
+        try
+        {
+            future.get(10, TimeUnit.SECONDS);
+            throw new AssertionError("Expected exceptional history completion");
+        }
+        catch (ExecutionException e)
+        {
+            assertTrue("Unexpected history failure: " + e.getCause(), type.isInstance(e.getCause()));
+            return e.getCause();
+        }
+    }
+
+    private static void await(CountDownLatch latch)
+    {
+        try
+        {
+            assertTrue(latch.await(20, TimeUnit.SECONDS));
+        }
+        catch (InterruptedException e)
+        {
+            Thread.currentThread().interrupt();
+            throw new AssertionError(e);
+        }
+    }
+}