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
The following commit(s) were added to refs/heads/trunk by this push:
new b6e924f8e5 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)
b6e924f8e5 is described below
commit b6e924f8e54d199bd16a177be64a163eccbba635
Author: Benedict Elliott Smith <[email protected]>
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 ++++++----
4 files changed, 35 insertions(+), 15 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..654a557c6a 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;
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);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
