natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Benedict Elliott Smith 2025-06-18 13:04:26 +01:00
parent 9a3fda57cd
commit 7ef2ccd4a1
5 changed files with 38 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 1ce7122e2e305f7510ec4c10c7587822c0549364
Subproject commit 2b80b0233f96464430058d70f7398df11066c3a6

26
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

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

10
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

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

12
src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
View File

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

6
test/distributed/org/apache/cassandra/distributed/test/accord/AccordProgressLogTest.java
View File

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