You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2025/06/19 11:41:00 UTC
(cassandra) 01/02: 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)
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 7ef2ccd4a1339a63d346523520fa2f9dae8352f5
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Wed Jun 18 13:04:26 2025 +0100
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 ++++++++++++++--------
.../cassandra/service/accord/AccordJournal.java | 10 ++++++++-
.../cassandra/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<UnfilteredRowIterator>
{
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);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org