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);