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

Improve GC testing; fix durability startup order 

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20767
This commit is contained in:
Alex Petrov 2025-07-14 19:18:27 +02:00
parent 54e39a90b0
commit 53eb2fdac8
7 changed files with 64 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 51f36f788b8095c0da9efaf1ea91bb9a7dd31181
Subproject commit e66a30cd944e902ad2749b5182e1f0d56903303f

2
src/java/org/apache/cassandra/journal/Compactor.java
View File

@ -70,7 +70,7 @@ public final class Compactor<K, V> implements Runnable, Shutdownable
{
Set<StaticSegment<K, V>> toCompact = new HashSet<>();
journal.segments().selectStatic(toCompact);
if (toCompact.size() < 2)
if (toCompact.isEmpty())
return;
try

1
src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java
View File

@ -96,7 +96,6 @@ public abstract class AbstractAccordSegmentCompactor<V> implements SegmentCompac
@Override
public Collection<StaticSegment<JournalKey, V>> compact(Collection<StaticSegment<JournalKey, V>> segments)
{
Invariants.require(segments.size() >= 2, () -> String.format("Can only compact 2 or more segments, but got %d", segments.size()));
logger.info("Compacting {} static segments: {}", segments.size(), segments);
// TODO (expected): this will be a large over-estimate. should make segments an sstable format and include cardinality estimation

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

@ -69,6 +69,13 @@ public class AccordDataStore implements DataStore
while (true)
{
Memtable memtable = cfs.getCurrentMemtable();
// If RX came when after a quiet period or if it raced with a previous memtable flush
if (memtable.isClean())
{
AccordDurableOnFlush.notify(cfs.metadata(), commandStore, reportOnSuccess);
break;
}
AccordDurableOnFlush onFlush = memtable.ensureFlushListener(FlushListenerKey.KEY, AccordDurableOnFlush::new);
if (onFlush != null && onFlush.add(commandStore.id(), reportOnSuccess))
break;

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

@ -36,7 +36,6 @@ import java.util.function.Supplier;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -376,17 +375,18 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
journal.onDurable(pointer, onFlush);
}
private static final JournalKey DURABLE_BEFORE_KEY = new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, 0);
@Override
public PersistentField.Persister<DurableBefore, DurableBefore> durableBeforePersister()
{
return new PersistentField.Persister<>()
{
@Override
public AsyncResult<?> persist(DurableBefore addDurableBefore, DurableBefore newDurableBefore)
public AsyncResult<?> persist(DurableBefore addValue, DurableBefore newValue)
{
AsyncResult.Settable<Void> result = AsyncResults.settable();
JournalKey key = new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, 0);
RecordPointer pointer = appendInternal(key, addDurableBefore);
RecordPointer pointer = appendInternal(DURABLE_BEFORE_KEY, addValue);
// TODO (required): what happens on failure?
journal.onDurable(pointer, () -> result.setSuccess(null));
return result;
@ -395,7 +395,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
@Override
public DurableBefore load()
{
DurableBeforeAccumulator accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.DURABLE_BEFORE, 0));
DurableBeforeAccumulator accumulator = readAll(DURABLE_BEFORE_KEY);
return accumulator.get();
}
};

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

@ -248,6 +248,10 @@ public class AccordService implements IAccordService, Shutdownable
instance = as;
replayJournal(as);
// Only enable durability scheduling _after_ we have fully replayed journal
as.configService.registerListener(as.node.durability());
as.node.durability().start();
}
@VisibleForTesting
@ -456,7 +460,6 @@ public class AccordService implements IAccordService, Shutdownable
Ints.checkedCast(getAccordShardDurabilityMaxSplits()),
Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS);
node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS);
node.durability().start();
state = State.STARTED;
}

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

@ -18,9 +18,16 @@
package org.apache.cassandra.fuzz.topology;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.test.log.FuzzTestBase;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
@ -34,10 +41,6 @@ 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.consensus.TransactionalMode;
import org.junit.Assert;
import org.junit.Test;
import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
@ -49,10 +52,13 @@ public class JournalGCTest extends FuzzTestBase
public void journalGCTest() throws Throwable
{
try (Cluster cluster = init(builder().withNodes(1)
.withConfig(cfg -> cfg.set("accord.gc_delay", "1s")
.set("accord.shard_durability_target_splits", "1")
.set("accord.shard_durability_cycle", "1s")
.set("accord.global_durability_cycle", "1s"))
.withConfig(cfg -> cfg.set("write_request_timeout", "2s")
.set("accord.expire_syncpoint", "1s*attempts<=300s")
.set("accord.retry_syncpoint", "1s*attempts")
.set("accord.shard_durability_target_splits", "5")
.set("accord.shard_durability_max_splits", "10")
.set("accord.shard_durability_cycle", "1s")
.set("accord.global_durability_cycle", "1s"))
.start()))
{
withRandom(rng -> {
@ -74,9 +80,23 @@ public class JournalGCTest extends FuzzTestBase
.pageSizeSelector(p -> InJvmDTestVisitExecutor.PageSizeSelector.NO_PAGING)
.build(schema, hb, cluster));
for (int pk = 0; pk < 500; pk++) {
for (int i = 0; i < 500; i++)
for (int pk = 0; pk <= 500; pk++) {
for (int i = 0; i < 100; i++)
history.insert(pk);
if (pk > 0 && pk % 100 == 0)
{
cluster.get(1).runOnInstance(() -> {
((AccordService) AccordService.instance()).journal().closeCurrentSegmentForTestingIfNonEmpty();
((AccordService) AccordService.instance()).journal().compactor().run();
});
}
if (pk > 0 && pk % 200 == 0)
{
ClusterUtils.stopUnchecked(cluster.get(1));
cluster.get(1).startup();
}
}
cluster.get(1).runOnInstance(() -> {
@ -84,34 +104,30 @@ public class JournalGCTest extends FuzzTestBase
((AccordService) AccordService.instance()).journal().compactor().run();
});
int before = cluster.get(1).callOnInstance(() -> {
Callable<Integer> countDiffs = cluster.get(1).callsOnInstance(() -> {
AtomicInteger a = new AtomicInteger();
((AccordService) AccordService.instance()).journal().forEach((v) -> {
if (v.type == JournalKey.Type.COMMAND_DIFF)
if (v.type == JournalKey.Type.COMMAND_DIFF &&
// Do not count syncpoints
!v.id.isSyncPoint())
a.incrementAndGet();
});
return a.get();
});
Thread.sleep(10_000);
cluster.get(1).runOnInstance(() -> {
Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL).forceMajorCompaction();
});
cluster.get(1).forceCompact("system_accord", "journal");
int after = cluster.get(1).callOnInstance(() -> {
AtomicInteger a = new AtomicInteger();
((AccordService) AccordService.instance()).journal().forEach((v) -> {
if (v.type == JournalKey.Type.COMMAND_DIFF)
a.incrementAndGet();
int after =-1;
for (int i = 0; i < 60; i++)
{
cluster.get(1).runOnInstance(() -> {
Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL).forceMajorCompaction();
});
return a.get();
});
Assert.assertTrue(String.format("%s should have been strictly smaller than %s", after, before), before > after);
Assert.assertEquals(0, after);
after = countDiffs.call();
if (after == 0)
return;
Thread.sleep(1000);
}
Assert.fail("Should have GC'd all in (way under) 60 cycles. Remaining: " + after);
});
}
}
}
}