From 537185e424653522a3707e7ac8d2ebbc1beed5a1 Mon Sep 17 00:00:00 2001 From: Mykyta Bozhenko <21245729+cheeeee@users.noreply.github.com> Date: Fri, 11 Sep 2026 18:50:21 -0400 Subject: [PATCH] Isolate best-effort compaction history writes with a bounded executor Move history mutations off compaction threads and shared maintenance schedulers. Return explicit completion, preserve failure inspection, and reject new diagnostic records when the bounded queue is full or shut down. Drain accepted history writes before storage teardown in production and distributed-instance lifecycles. CASSANDRA-19576 Generated-by: Claude (Anthropic) --- CHANGES.txt | 1 + .../cassandra/db/CompactionHistoryWriter.java | 70 ++++++++ .../apache/cassandra/db/SystemKeyspace.java | 48 ++++-- .../cassandra/service/StorageService.java | 1 + .../cassandra/distributed/impl/Instance.java | 3 + .../test/CompactionHistoryShutdownTest.java | 57 +++++++ .../db/CompactionHistoryWriterTest.java | 150 ++++++++++++++++++ .../cassandra/db/SystemKeyspaceTest.java | 28 +++- .../db/compaction/CompactionTaskTest.java | 3 + .../tools/nodetool/CompactionHistoryTest.java | 6 + 10 files changed, 353 insertions(+), 14 deletions(-) create mode 100644 src/java/org/apache/cassandra/db/CompactionHistoryWriter.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/CompactionHistoryShutdownTest.java create mode 100644 test/unit/org/apache/cassandra/db/CompactionHistoryWriterTest.java diff --git a/CHANGES.txt b/CHANGES.txt index aa356a2da163..dab0a3254608 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/db/CompactionHistoryWriter.java b/src/java/org/apache/cassandra/db/CompactionHistoryWriter.java new file mode 100644 index 000000000000..1c603aeb187b --- /dev/null +++ b/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 builder) + { + executor = builder.withQueueLimit(queueLimit) + .withRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()) + .build(); + } + + Future 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); + } +} diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 8010d5b799c6..862596eca1a1 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/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 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 rows = ImmutableMap.copyOf(rowsMerged); + Map 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 diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index ba0edbdd1103..43216b003c8c 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/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 diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java index 98ea380b63cc..8ce802d6dcf6 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@ -977,6 +977,9 @@ public Future 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), diff --git a/test/distributed/org/apache/cassandra/distributed/test/CompactionHistoryShutdownTest.java b/test/distributed/org/apache/cassandra/distributed/test/CompactionHistoryShutdownTest.java new file mode 100644 index 000000000000..95ee4556ea00 --- /dev/null +++ b/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); + } + } +} diff --git a/test/unit/org/apache/cassandra/db/CompactionHistoryWriterTest.java b/test/unit/org/apache/cassandra/db/CompactionHistoryWriterTest.java new file mode 100644 index 000000000000..0beb77a4b1df --- /dev/null +++ b/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 writes = new ArrayList<>(); + try + { + Future first = writer.submit(() -> { + started.countDown(); + await(release); + writes.add(1); + }); + assertTrue(started.await(10, TimeUnit.SECONDS)); + Future second = writer.submit(() -> writes.add(2)); + Future 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 writes = new ArrayList<>(); + try + { + Future first = writer.submit(() -> { + started.countDown(); + await(release); + writes.add(1); + }); + assertTrue(started.await(10, TimeUnit.SECONDS)); + Future 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 future, Class 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); + } + } +} diff --git a/test/unit/org/apache/cassandra/db/SystemKeyspaceTest.java b/test/unit/org/apache/cassandra/db/SystemKeyspaceTest.java index 9e97f7517860..f3c375b67b68 100644 --- a/test/unit/org/apache/cassandra/db/SystemKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/SystemKeyspaceTest.java @@ -27,6 +27,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.TimeUnit; import org.junit.BeforeClass; import org.junit.Test; @@ -167,7 +168,7 @@ public void testPersistLocalMetadata() } @Test - public void testCompactionHistory() + public void testCompactionHistory() throws Exception { String ks = "test_ks"; String cf = "test_cf"; @@ -188,7 +189,7 @@ public void testCompactionHistory() 500, rowsMerged, propertiesWithType - ); + ).get(10, TimeUnit.SECONDS); UntypedResultSet result = executeInternal("SELECT compaction_properties FROM system.compaction_history WHERE keyspace_name=? AND columnfamily_name=? ALLOW FILTERING", ks, cf); @@ -198,6 +199,29 @@ public void testCompactionHistory() assertEquals(compactionType, resProps.get("compaction_type")); } + @Test(timeout = 30000) + public void testCompactionHistoryDefensiveCopy() throws Exception + { + TimeUUID id = TimeUUID.Generator.nextTimeUUID(); + Map rows = new HashMap<>(); + rows.put(1, 100L); + Map properties = new HashMap<>(); + properties.put("strategy", "STCS"); + + SystemKeyspace.updateCompactionHistory(id, "async_history_ks", "async_history_cf", + System.currentTimeMillis(), 1000, 500, rows, properties) + .get(10, TimeUnit.SECONDS); + // Mutating the caller's maps after the call must not affect the stored record. + rows.clear(); + properties.clear(); + + UntypedResultSet.Row stored = executeInternal("SELECT rows_merged, compaction_properties FROM system.compaction_history WHERE id=?", id).one(); + assertEquals(Collections.singletonMap(1, 100L), + stored.getMap("rows_merged", org.apache.cassandra.db.marshal.Int32Type.instance, org.apache.cassandra.db.marshal.LongType.instance)); + assertEquals(Collections.singletonMap("strategy", "STCS"), + stored.getMap("compaction_properties", org.apache.cassandra.db.marshal.UTF8Type.instance, org.apache.cassandra.db.marshal.UTF8Type.instance)); + } + private String getOlderVersionString() { String version = FBUtilities.getReleaseVersionString(); diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java index ab529daa5bd4..37f12d8c7259 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java @@ -113,6 +113,9 @@ public void testTaskIdIsPersistedInCompactionHistory() CompactionTask task = new CompactionTask(cfs, txn, 0); task.execute(CompactionManager.instance.active); } + // Compaction completion precedes visibility of its best-effort history record. + Util.spinAssertEquals(1, () -> QueryProcessor.executeInternal(format("SELECT id FROM system.%s WHERE id = %s", + SystemKeyspace.COMPACTION_HISTORY, id)).size()); UntypedResultSet rows = QueryProcessor.executeInternal(format("SELECT id, compaction_properties FROM system.%s where id = %s", SystemKeyspace.COMPACTION_HISTORY, diff --git a/test/unit/org/apache/cassandra/tools/nodetool/CompactionHistoryTest.java b/test/unit/org/apache/cassandra/tools/nodetool/CompactionHistoryTest.java index b9ee543a4358..59f13a6c167d 100644 --- a/test/unit/org/apache/cassandra/tools/nodetool/CompactionHistoryTest.java +++ b/test/unit/org/apache/cassandra/tools/nodetool/CompactionHistoryTest.java @@ -36,6 +36,7 @@ import org.junit.runners.Parameterized.Parameter; import org.junit.runners.Parameterized.Parameters; +import org.apache.cassandra.Util; import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; @@ -124,6 +125,11 @@ private void compactionHistoryResultVerify(String keyspace, String table, Map org.apache.cassandra.cql3.QueryProcessor.executeInternal( + "SELECT id FROM system.compaction_history WHERE keyspace_name=? AND columnfamily_name=? ALLOW FILTERING", + keyspace, table).size()); + ToolResult toolHistory = invokeNodetool("compactionhistory"); toolHistory.assertOnCleanExit(); assertCompactionHistoryOutPut(toolHistory, keyspace, table, properties, compType);