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

Fix DurableBefore updates: 

- DurableBefore.fullyContainedIn should compare txnId
Also fix:
 - Messages with nullable state should check if they are cancelled after reading the state but before using it
 - Don't snapshot during replay
 - CFK.updateRedundantBefore incorrectly updating mayExecute and sharing byId in some cases
 - Refine GC with global durability bound
 - Fix JournalGCTest

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20786
This commit is contained in:
Benedict Elliott Smith 2025-07-24 18:35:29 +01:00
parent fbe12e6291
commit 49b99c9319
4 changed files with 34 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit f40826db9571af286a689deada1613ffc872e5f8
Subproject commit 090cb49bfb52357fee0f554d5ba0f0b012195280

5
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -505,6 +505,9 @@ public class AccordCommandStore extends CommandStore
@Override
protected void ensureDurable(Ranges ranges, RedundantBefore onCommandStoreDurable)
{
if (!CommandsForKey.reportLinearizabilityViolations())
return;
long reportId = nextDurabilityLoggingId.incrementAndGet();
logger.debug("{} awaiting local metadata durability for {} ({})", this, ranges, reportId);
executor().afterSubmittedAndConsequences(() -> {
@ -515,8 +518,6 @@ public class AccordCommandStore extends CommandStore
@Override public void run() { decrement(); }
}
Map<RoutingKey, String> saving = new TreeMap<>();
Map<RoutingKey, String> excluding = new TreeMap<>();
Ready ready = new Ready();
try (ExclusiveCaches caches = lockCaches())
{

4
src/java/org/apache/cassandra/service/accord/AccordDataStore.java
View File

@ -26,6 +26,7 @@ import accord.local.CommandStore;
import accord.local.Node;
import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.local.cfk.CommandsForKey;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.SyncPoint;
@ -53,6 +54,9 @@ public class AccordDataStore implements DataStore
*/
public void ensureDurable(CommandStore commandStore, Ranges ranges, RedundantBefore reportOnSuccess)
{
if (!CommandsForKey.reportLinearizabilityViolations())
return;
logger.debug("{} awaiting local data durability of {}", commandStore, ranges);
ColumnFamilyStore prev = null;
for (Range range : ranges)

37
test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java
View File

@ -20,10 +20,12 @@ package org.apache.cassandra.fuzz.topology;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Assert;
import org.junit.Test;
import accord.primitives.TxnId;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
@ -40,8 +42,10 @@ import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.JournalKey;
import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.service.consensus.TransactionalMode;
import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
public class JournalGCTest extends FuzzTestBase
@ -104,29 +108,40 @@ public class JournalGCTest extends FuzzTestBase
((AccordService) AccordService.instance()).journal().compactor().run();
});
Callable<Integer> countDiffs = cluster.get(1).callsOnInstance(() -> {
AtomicInteger a = new AtomicInteger();
String maximumId = cluster.get(1).callOnInstance(() -> {
AtomicReference<TxnId> a = new AtomicReference<>();
((AccordService) AccordService.instance()).journal().forEach((v) -> {
if (v.type == JournalKey.Type.COMMAND_DIFF &&
// Do not count syncpoints
!v.id.isSyncPoint())
if (v.type == JournalKey.Type.COMMAND_DIFF && (a.get() == null || v.id.compareTo(a.get()) > 0))
a.set(v.id);
});
return a.get() == null ? "" : a.get().toString();
});
Callable<Integer> countDiffs = () -> cluster.get(1).applyOnInstance(maxIdStr -> {
AtomicInteger a = new AtomicInteger();
TxnId maxId = TxnId.parse(maxIdStr);
((AccordService) AccordService.instance()).journal().forEach((v) -> {
if (v.type == JournalKey.Type.COMMAND_DIFF && v.id.compareTo(maxId) <= 0)
a.incrementAndGet();
});
return a.get();
});
}, maximumId);
int after =-1;
for (int i = 0; i < 60; i++)
int maxCycles = 3;
for (int i = 0; i < maxCycles; i++)
{
cluster.get(1).runOnInstance(() -> {
cluster.get(1).acceptOnInstance((ks, tbl) -> {
Keyspace.open(ks).getColumnFamilyStore(tbl).forceBlockingFlush(UNIT_TESTS);
Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.COMMANDS_FOR_KEY).forceBlockingFlush(UNIT_TESTS);
Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL).forceMajorCompaction();
});
}, schema.keyspace, schema.table);
after = countDiffs.call();
if (after == 0)
return;
Thread.sleep(1000);
Thread.sleep(10000);
}
Assert.fail("Should have GC'd all in (way under) 60 cycles. Remaining: " + after);
Assert.fail("Should have GC'd all in (way under) " + maxCycles + " cycles. Remaining: " + after);
});
}
}