You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2025/06/29 11:59:14 UTC
(cassandra) branch trunk updated: 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 imp [...]
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 0a7797ca85 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 [...]
0a7797ca85 is described below
commit 0a7797ca85d78052638d47b5e1b3a5886839333d
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Fri Jun 27 22:08:12 2025 +0100
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
---
modules/accord | 2 +-
.../cassandra/service/accord/AccordCache.java | 3 ++
.../cassandra/service/accord/AccordJournal.java | 10 +----
.../service/accord/AccordSyncPropagator.java | 4 +-
.../cassandra/service/accord/JournalKey.java | 3 +-
.../org/apache/cassandra/utils/btree/BTree.java | 6 +++
.../distributed/test/accord/AccordLoadTest.java | 18 +++++---
.../service/accord/AccordJournalBurnTest.java | 49 +++++++++++++++++++++-
8 files changed, 75 insertions(+), 20 deletions(-)
diff --git a/modules/accord b/modules/accord
index aa08b04e82..555337a7d4 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit aa08b04e8235d9c0b7e2bbe3bbe8773184bbbe46
+Subproject commit 555337a7d41158f74033818facf94fed6904bf5a
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCache.java b/src/java/org/apache/cassandra/service/accord/AccordCache.java
index 704ed31881..b7340d3a66 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCache.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCache.java
@@ -1132,6 +1132,9 @@ public class AccordCache implements CacheSize
if (value.isEmpty())
return null;
+ if (value.isLoadingPruned())
+ return value;
+
return Serialize.toBytesWithoutKey(value.maximalPrune());
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index db64d55da2..72bca48480 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -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);
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
index 66ae868231..cf9e5db835 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
@@ -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;
}
diff --git a/src/java/org/apache/cassandra/service/accord/JournalKey.java b/src/java/org/apache/cassandra/service/accord/JournalKey.java
index b299f91613..9bd42ebaf1 100644
--- a/src/java/org/apache/cassandra/service/accord/JournalKey.java
+++ b/src/java/org/apache/cassandra/service/accord/JournalKey.java
@@ -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;
}
diff --git a/src/java/org/apache/cassandra/utils/btree/BTree.java b/src/java/org/apache/cassandra/utils/btree/BTree.java
index 201554512c..08372f26a5 100644
--- a/src/java/org/apache/cassandra/utils/btree/BTree.java
+++ b/src/java/org/apache/cassandra/utils/btree/BTree.java
@@ -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
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
index 2a093887b6..66e188df5c 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java
@@ -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)
diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
index 50cc6c65de..2101ba8d14 100644
--- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
+++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
@@ -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)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org