From 7ef2ccd4a1339a63d346523520fa2f9dae8352f5 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Wed, 18 Jun 2025 13:04:26 +0100 Subject: [PATCH] Erase should consistently erase all content Also Fix: - slowCoordinatorDelay should bound its start point, and log more detail if its expectations are breached - Erased SaveStatus can be reported for all queried owned participants on a replica (the saved participants have been erased) patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20722 --- modules/accord | 2 +- .../db/compaction/CompactionIterator.java | 26 ++++++++++++------- .../service/accord/AccordJournal.java | 10 ++++++- .../service/accord/api/AccordAgent.java | 12 ++++++--- .../test/accord/AccordProgressLogTest.java | 6 ++--- 5 files changed, 38 insertions(+), 18 deletions(-) diff --git a/modules/accord b/modules/accord index 1ce7122e2e..2b80b0233f 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 1ce7122e2e305f7510ec4c10c7587822c0549364 +Subproject commit 2b80b0233f96464430058d70f7398df11066c3a6 diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 53b4d57f2e..e95b100a3f 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -35,6 +35,7 @@ import com.google.common.collect.Ordering; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import accord.impl.CommandChange; import accord.local.Cleanup; import accord.local.DurableBefore; import accord.local.RedundantBefore; @@ -109,6 +110,7 @@ import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement; import org.apache.cassandra.utils.TimeUUID; +import static accord.local.Cleanup.ERASE; import static accord.local.Cleanup.Input.PARTIAL; import static accord.local.Cleanup.NO; import static com.google.common.base.Preconditions.checkState; @@ -117,6 +119,7 @@ import static java.util.concurrent.TimeUnit.MINUTES; import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy; import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging; import static org.apache.cassandra.service.accord.AccordKeyspace.CFKAccessor; +import static org.apache.cassandra.service.accord.AccordKeyspace.JournalColumns.getJournalKey; /** * Merge multiple iterators over the content of sstable into a "compacted" iterator. @@ -880,7 +883,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte @Override protected void beginPartition(UnfilteredRowIterator partition) { - key = AccordKeyspace.JournalColumns.getJournalKey(partition.partitionKey()); + key = getJournalKey(partition.partitionKey()); if (compactor == null || compactor.serializer != key.type.serializer) { switch (key.type) @@ -895,15 +898,12 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte compactor = new AccordMergingCompactor(key.type.serializer, userVersion); } } - compactor.reset(key); + compactor.reset(key, partition); } @Override protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition) { - if (!partition.hasNext()) - return partition; - try { beginPartition(partition); @@ -941,7 +941,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte this.serializer = serializer; } - abstract void reset(JournalKey key); + abstract void reset(JournalKey key, UnfilteredRowIterator partition); abstract void collect(JournalKey key, Row row, ByteBuffer bytes, Version userVersion) throws IOException; abstract UnfilteredRowIterator result(JournalKey journalKey, DecoratedKey partitionKey) throws IOException; } @@ -962,7 +962,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte } @Override - void reset(JournalKey key) + void reset(JournalKey key, UnfilteredRowIterator partition) { builder.reset(key); lastDescriptor = -1; @@ -1060,13 +1060,15 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte } @Override - void reset(JournalKey key) + void reset(JournalKey key, UnfilteredRowIterator partition) { mainBuilder.reset(key); reuseEntries.addAll(entries); for (int i = 0; i < entries.size() ; ++i) entries.get(i).clear(); entries.clear(); + if (!partition.partitionLevelDeletion().isLive()) + mainBuilder.addCleanup(false, ERASE); } @Override @@ -1103,7 +1105,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte case EXPUNGE: return null; case ERASE: - return PartitionUpdate.fullPartitionDelete(AccordKeyspace.Journal, partitionKey, Long.MAX_VALUE, nowInSec).unfilteredIterator(); + return erase(partitionKey); case TRUNCATE: case TRUNCATE_WITH_OUTCOME: @@ -1136,8 +1138,14 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte } return newVersion.build().unfilteredIterator(); } + + private UnfilteredRowIterator erase(DecoratedKey partitionKey) + { + return PartitionUpdate.fullPartitionDelete(AccordKeyspace.Journal, partitionKey, Long.MAX_VALUE, nowInSec).unfilteredIterator(); + } } + private static class AbortableUnfilteredPartitionTransformation extends Transformation { private final AbortableUnfilteredRowTransformation abortableIter; diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index f98061f887..d7c507e34f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -588,7 +588,15 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier ResultSerializers.result.serialize(command.result(), out); break; case CLEANUP: - throw new IllegalStateException(); + Cleanup cleanup; + switch (command.saveStatus()) + { + default: throw new UnhandledEnum(command.saveStatus()); + case Erased: cleanup = Cleanup.ERASE; break; + case Invalidated: cleanup = Cleanup.INVALIDATE; break; + } + out.writeByte(cleanup.ordinal()); + break; } iterable = unsetIterable(field, iterable); diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index 949f7bbf86..9f07e351ab 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -223,18 +223,22 @@ public class AccordAgent implements Agent Command command = safeCommand.current(); Invariants.nonNull(command); - Timestamp mostRecentAttempt = Timestamp.max(command.txnId(), command.promised()); RoutingKey homeKey = command.route().homeKey(); Shard shard = node.topology().forEpochIfKnown(homeKey, command.txnId().epoch()); // TODO (expected): make this a configurable calculation on normal request latencies (like ContentionStrategy) long oneSecond = SECONDS.toMicros(1L); - long startTime = mostRecentAttempt.hlc() + recover(txnId).computeWait(retryCount, MICROSECONDS); + long mostRecentStart = Math.max(command.txnId().hlc(), command.promised().hlc()); + long waitMicros = recover(txnId).computeWait(retryCount, MICROSECONDS); + long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis()); + Invariants.expect(mostRecentStart <= nowMicros + SECONDS.toMicros(1L), "max(%s,%s)>%d", command.txnId(), command.promised(), nowMicros); + long startTime = mostRecentStart + waitMicros; + if (startTime < nowMicros) + startTime = nowMicros + waitMicros/2; startTime = nonClashingStartTime(startTime, shard == null ? null : shard.nodes, node.id(), oneSecond, random); - long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis()); long delayMicros = Math.max(1, startTime - nowMicros); - Invariants.require(delayMicros < TimeUnit.HOURS.toMicros(1L)); + Invariants.require(delayMicros < TimeUnit.HOURS.toMicros(1L), "unexpectedly long coordination recovery delay proposed: %d (start %d, now %d)", delayMicros, startTime, nowMicros, command.txnId(), command.promised()); return units.convert(delayMicros, MICROSECONDS); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java index 6bbbce6a52..3f4163ddb2 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java @@ -46,7 +46,7 @@ public class AccordProgressLogTest extends TestBaseImpl .withoutVNodes() .withConfig(c -> c.with(Feature.NETWORK) .set("accord.enabled", "true") - .set("accord.recover_txn", "1s")) + .set("accord.recover_txn", "100ms")) .start())) { cluster.schemaChange("CREATE KEYSPACE ks WITH replication={'class':'SimpleStrategy', 'replication_factor': 3}"); @@ -59,7 +59,7 @@ public class AccordProgressLogTest extends TestBaseImpl AtomicLong recoveryStartedAt = new AtomicLong(); Semaphore waitForRecovery = new Semaphore(0); IMessageFilters.Filter recovery = cluster.filters().outbound().messagesMatching((from, to, message) -> { - if (message.verb() == Verb.ACCORD_BEGIN_RECOVER_RSP.id) + if (message.verb() == Verb.ACCORD_BEGIN_RECOVER_REQ.id) { recoveryStartedAt.compareAndSet(0, System.nanoTime()); waitForRecovery.release(); @@ -75,7 +75,7 @@ public class AccordProgressLogTest extends TestBaseImpl waitForRecovery.acquire(); long timeDeltaMillis = TimeUnit.NANOSECONDS.toMillis(recoveryStartedAt.get() - coordinationStartedAt); - Assert.assertTrue("Recovery started in " + timeDeltaMillis + "ms", timeDeltaMillis >= 1000); + Assert.assertTrue("Recovery started in " + timeDeltaMillis + "ms", timeDeltaMillis >= 100); Assert.assertTrue("Recovery started in " + timeDeltaMillis + "ms", timeDeltaMillis <= 5000); } }