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

Fix: 

- Cannot shrink CommandsForKey if loading pruned
 - NoSpamLogger to suppress AccordSyncPropagator repeats
 - Minor performance improvement to BTree.Subtraction
 - EphemeralReads should retry in a later epoch if replication factor changes
 - Fix no local epoch for NotAccept
 - Invalidate a command with no route but for which we know we own some key that has applied locally
 - Don't pre-merge existing DurableBefore, to reduce duplicate/redundant persistence
 - Misc purging bugs and improve testing of purging

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20739
This commit is contained in:
Benedict Elliott Smith 2025-06-27 22:08:12 +01:00
parent 78d6892dfb
commit 0a7797ca85
8 changed files with 75 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit aa08b04e8235d9c0b7e2bbe3bbe8773184bbbe46
Subproject commit 555337a7d41158f74033818facf94fed6904bf5a

3
src/java/org/apache/cassandra/service/accord/AccordCache.java
View File

@ -1132,6 +1132,9 @@ public class AccordCache implements CacheSize
if (value.isEmpty())
return null;
if (value.isLoadingPruned())
return value;
return Serialize.toBytesWithoutKey(value.maximalPrune());
}

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

@ -224,15 +224,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
public Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
Builder builder = load(commandStoreId, txnId);
Cleanup cleanup = builder.maybeCleanup(true, FULL, redundantBefore, durableBefore);
switch (cleanup)
{
case ERASE:
return Command.Truncated.erased(txnId);
case EXPUNGE:
return null;
}
builder.maybeCleanup(true, FULL, redundantBefore, durableBefore);
return builder.construct(redundantBefore);
}

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

@ -55,6 +55,7 @@ import org.apache.cassandra.service.accord.serializers.KeySerializers;
import org.apache.cassandra.service.accord.serializers.TopologySerializers;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.CollectionSerializers;
import org.apache.cassandra.utils.NoSpamLogger;
/**
* Receives information about closed, retired ranges, and about sync completion, and
@ -66,6 +67,7 @@ import org.apache.cassandra.utils.CollectionSerializers;
public class AccordSyncPropagator
{
private static final Logger logger = LoggerFactory.getLogger(AccordSyncPropagator.class);
private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES);
public static final IVerbHandler<Notification> verbHandler = message -> {
if (!AccordService.isSetup())
@ -352,7 +354,7 @@ public class AccordSyncPropagator
cb.onResponse(msg.responseWith(SimpleReply.Ok));
return true;
}
logger.warn("Node{} is not alive, unable to notify of {}", to, notification);
noSpamLogger.warn("Node{} is not alive, unable to notify of {}", to, notification);
scheduler.schedule(() -> notify(to, notification), 1, TimeUnit.MINUTES);
return false;
}

3
src/java/org/apache/cassandra/service/accord/JournalKey.java
View File

@ -234,7 +234,8 @@ public final class JournalKey
boolean equals(JournalKey other)
{
return this.id.equals(other.id) &&
return other != null &&
this.id.equals(other.id) &&
this.type == other.type &&
this.commandStoreId == other.commandStoreId;
}

6
src/java/org/apache/cassandra/utils/btree/BTree.java
View File

@ -4029,6 +4029,12 @@ public class BTree
if (leaf().count == 0 && upos == 0)
{
int prev = 0;
if (!remove.hasNext() || comparator.compare((K)unode[usz - 1], remove.peek()) < 0)
{
// short-circuit common case of removal not intersecting the current node, by comparing with last key
upos = usz;
}
while (upos < usz)
{
// fast path - buffer is empty and input unconsumed, so may be able to propagate original

18
test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
View File

@ -70,8 +70,8 @@ public class AccordLoadTest extends AccordTestBase
// AccordTestBase.setupCluster(builder -> builder, 3);
AccordTestBase.setupCluster(builder -> builder.withConfig(config -> config
.with(Feature.NETWORK, Feature.GOSSIP)
.set("accord.shard_durability_target_splits", "64")
.set("accord.shard_durability_cycle", "5m")
.set("accord.shard_durability_target_splits", "32")
.set("accord.shard_durability_cycle", "1m")
// .set("accord.ephemeral_read_enabled", "true")
.set("accord.gc_delay", "30s")), 3);
}
@ -103,8 +103,9 @@ public class AccordLoadTest extends AccordTestBase
// final int flushInterval = 50_000;
final int flushInterval = 500;
final int compactionPeriodSeconds = 0;
// final int restartInterval = 100_000;
final int restartInterval = Integer.MAX_VALUE;
int restartInterval = 30_000;
final int restartDecay = 2;
// final int restartInterval = Integer.MAX_VALUE;
final int batchSizeLimit = 200;
final long batchTime = TimeUnit.SECONDS.toNanos(10);
final int concurrency = 100;
@ -176,7 +177,8 @@ public class AccordLoadTest extends AccordTestBase
int index = 1 + random.nextInt(cluster.size());
logger.info("Picking new coordinator ... {}", index);
coordinator = cluster.coordinator(index);
break;
if (cluster.get(index).callOnInstance(() -> AccordService.started()))
break;
}
catch (Throwable t) { logger.info("Failed to select coordinator", t); }
}
@ -211,7 +213,8 @@ public class AccordLoadTest extends AccordTestBase
try
{
i.runOnInstance(() -> {
((AccordService) AccordService.instance()).journal().closeCurrentSegmentForTestingIfNonEmpty();
if (AccordService.started())
((AccordService) AccordService.instance()).journal().closeCurrentSegmentForTestingIfNonEmpty();
});
}
catch (Throwable t)
@ -224,6 +227,7 @@ public class AccordLoadTest extends AccordTestBase
if ((nextRestartAt -= batchSize) <= 0)
{
nextRestartAt += restartInterval;
restartInterval = Math.max(restartInterval, restartInterval * restartDecay);
int nodeIdx = 1 + random.nextInt(cluster.size());
restartExecutor.submit(() -> {
System.out.printf("restarting node %d...\n", nodeIdx);
@ -231,6 +235,8 @@ public class AccordLoadTest extends AccordTestBase
{
cluster.get(nodeIdx).shutdown().get();
cluster.get(nodeIdx).startup();
while (!cluster.get(nodeIdx).callOnInstance(() -> AccordService.started()))
Thread.sleep(1000);
return null;
}
catch (InterruptedException | ExecutionException e)

49
test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
View File

@ -24,7 +24,10 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
@ -39,6 +42,8 @@ import accord.burn.SimulationException;
import accord.impl.TopologyFactory;
import accord.impl.basic.Cluster;
import accord.impl.basic.RandomDelayQueue;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.DurableBefore;
import accord.local.Node;
@ -63,6 +68,7 @@ import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.journal.Journal;
import org.apache.cassandra.journal.SegmentCompactor;
import org.apache.cassandra.journal.StaticSegment;
import org.apache.cassandra.journal.TestParams;
@ -78,6 +84,7 @@ import org.apache.cassandra.service.accord.serializers.ResultSerializers;
import org.apache.cassandra.service.accord.serializers.TopologySerializers;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.tools.FieldUtil;
import org.apache.cassandra.utils.CloseableIterator;
import static accord.impl.PrefixedIntHashKey.ranges;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
@ -129,7 +136,7 @@ public class AccordJournalBurnTest extends BurnTestBase
public void testOne()
{
long seed = System.nanoTime();
int operations = 5000;
int operations = 1000;
logger.info("Seed: {}", seed);
Cluster.trace.trace("Seed: {}", seed);
@ -243,9 +250,11 @@ public class AccordJournalBurnTest extends BurnTestBase
return new DefaultCompactionWriter(cfs, directories, transaction, nonExpiredSSTables, false, 0);
}
int counter;
@Override
public void purge(CommandStores commandStores, EpochSupplier minEpoch)
{
++counter;
this.journal.closeCurrentSegmentForTestingIfNonEmpty();
this.journal.runCompactorForTesting();
@ -272,6 +281,7 @@ public class AccordJournalBurnTest extends BurnTestBase
return;
List<ISSTableScanner> scanners = selected.stream().map(SSTableReader::getScanner).collect(Collectors.toList());
TreeMap<JournalKey, Command> before = read(commandStores);
Collection<SSTableReader> newSStables;
try (LifecycleTransaction txn = cfs.getTracker().tryModify(selected, OperationType.COMPACTION);
CompactionController controller = new CompactionController(cfs, selected, 0);
@ -298,10 +308,45 @@ public class AccordJournalBurnTest extends BurnTestBase
throw new RuntimeException(e);
}
}
TreeMap<JournalKey, Command> after = read(commandStores);
for (Map.Entry<JournalKey, Command> e : before.entrySet())
{
Command b = e.getValue();
Command a = after.get(e.getKey());
Invariants.require(Objects.equals(a, b));
}
if (before.size() != after.size())
{
for (Map.Entry<JournalKey, Command> e : after.entrySet())
Invariants.require(null != before.get(e.getKey()));
Invariants.require(false);
}
Invariants.require(!orig.equals(cfs.getLiveSSTables()));
}
private TreeMap<JournalKey, Command> read(CommandStores commandStores)
{
TreeMap<JournalKey, Command> result = new TreeMap<>(JournalKey.SUPPORT::compare);
try (CloseableIterator<Journal.KeyRefs<JournalKey>> iter = journalTable.keyIterator())
{
JournalKey prev = null;
while (iter.hasNext())
{
Journal.KeyRefs<JournalKey> ref = iter.next();
if (ref.key().type != JournalKey.Type.COMMAND_DIFF)
continue;
JournalKey key = ref.key();
if (key.equals(prev)) continue;
CommandStore commandStore = commandStores.forId(ref.key().commandStoreId);
Command command = loadCommand(key.commandStoreId, key.id, commandStore.unsafeGetRedundantBefore(), commandStore.durableBefore());
if (command != null)
result.put(key, command);
prev = key;
}
}
return result;
}
@Override
public void replay(CommandStores commandStores)