From b91731b5a3658230fe0cd0fab04e405fb61aaae3 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Wed, 3 Sep 2025 14:51:52 +0100 Subject: [PATCH] All pending topology notifications must be propagated to a newly published epoch, not only the pending notification for that epoch Also Fix: - JournalAndTableKeyIterator not merging in consistent order, which can lead to replaying topologies in non-ascending order - Invariant failure when reporting topologies if fetchTopologies on startup yields a gap - removeRedundantMissing could erroneously remove missing dependencies occurring after the new appliedBeforeIndex - Invariant failure for some propagate calls supplying 'wrong' addRoute - Disambiguate nextTxnId and nextTxnIdWithDefaultFlags Also Improve: - LocalLog should not use NoSuchElementException for control flow (instead use ConcurrentSkipListMap) patch by Benedict; reviewed by Aleksey Yeschenko for CASSANDRA-20886 - JournalAndTableKeyIterator not merging in consistent order --- modules/accord | 2 +- .../service/accord/AccordJournal.java | 23 ++++++++++-- .../service/accord/AccordJournalTable.java | 2 +- .../service/accord/AccordService.java | 37 +++++-------------- .../service/accord/IAccordService.java | 10 ++--- .../accord/interop/AccordInteropAdapter.java | 1 + .../apache/cassandra/tcm/log/LocalLog.java | 36 ++++++------------ .../db/virtual/AccordDebugKeyspaceTest.java | 12 +++--- .../AccordConfigurationServiceTest.java | 7 ++-- 9 files changed, 58 insertions(+), 72 deletions(-) diff --git a/modules/accord b/modules/accord index 48061a521c..c3a62d77ba 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 48061a521cde7a085fe7a5503023c8ef8b687a89 +Subproject commit c3a62d77ba332f4a01220709c040af86dd3acf6a diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 7369b5e280..68a3b6c138 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -370,9 +370,10 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier } @Override - public CloseableIterator replayTopologies() + public List replayTopologies() { - return new CloseableIterator<>() + List images = new ArrayList<>(); + try (CloseableIterator iter = new CloseableIterator<>() { final CloseableIterator> iter = journalTable.keyIterator(topologyUpdateKey(0L), topologyUpdateKey(Timestamp.MAX_EPOCH)); @@ -388,6 +389,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier public TopologyUpdate next() { Journal.KeyRefs ref = iter.next(); + System.out.println(ref.key()); Accumulator read = readAll(ref.key()); if (read.accumulated.kind() == Kind.NoOp) prev = read.accumulated.asImage(Invariants.nonNull(prev.getUpdate())); @@ -403,7 +405,22 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier { iter.close(); } - }; + }) + { + TopologyUpdate prev = null; + while (iter.hasNext()) + { + TopologyUpdate next = iter.next(); + Invariants.require(prev == null || next.global.epoch() > prev.global.epoch()); + // Due to partial compaction, we can clean up only some of the old epochs, creating gaps. We skip these epochs here. + if (prev != null && next.global.epoch() > prev.global.epoch() + 1) + images.clear(); + + images.add(next); + prev = next; + } + } + return images; } @Override diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java index d461f6a8dc..984ab7ed62 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java @@ -567,7 +567,7 @@ public class AccordJournalTable implements RangeSearche return journalIterator.next(); } - return cmp > 0 ? new Journal.KeyRefs<>(tableIterator.next()) : journalIterator.next(); + return cmp < 0 ? new Journal.KeyRefs<>(tableIterator.next()) : journalIterator.next(); } public void close() diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 2549cc5b3d..57ac17660c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -128,7 +128,6 @@ import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.listeners.ChangeListener; import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.transport.Dispatcher; -import org.apache.cassandra.utils.CloseableIterator; import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.concurrent.AsyncPromise; @@ -144,7 +143,6 @@ import static accord.local.durability.DurabilityService.SyncRemote.All; import static accord.messages.SimpleReply.Ok; import static accord.primitives.Txn.Kind.ExclusiveSyncPoint; import static accord.primitives.Txn.Kind.Write; -import static accord.primitives.TxnId.Cardinality.cardinality; import static accord.topology.TopologyManager.TopologyRange; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static java.util.concurrent.TimeUnit.SECONDS; @@ -203,6 +201,7 @@ public class AccordService implements IAccordService, Shutdownable @Override public synchronized void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot) { + logger.debug("Saving epoch {} to deliver after startup", next.epoch); items.add(next); } @@ -428,29 +427,11 @@ public class AccordService implements IAccordService, Shutdownable ClusterMetadata metadata = ClusterMetadata.current(); configService.updateMapping(metadata); - List images = new ArrayList<>(); - - TopologyUpdate prev = null; - // Collect locally known topologies - try (CloseableIterator iter = journal.replayTopologies()) - { - while (iter.hasNext()) - { - TopologyUpdate next = iter.next(); - // Due to partial compaction, we can clean up only some of the old epochs, creating gaps. We skip these epochs here. - if (prev != null && next.global.epoch() > prev.global.epoch() + 1) - images.clear(); - - images.add(next); - prev = next; - } - } + List images = journal.replayTopologies(); // Instantiate latest topology from the log, if known - if (prev != null) - { - node.commandStores().initializeTopologyUnsafe(prev); - } + if (!images.isEmpty()) + node.commandStores().initializeTopologyUnsafe(images.get(images.size() - 1)); // Replay local epochs for (TopologyUpdate image : images) @@ -494,8 +475,8 @@ public class AccordService implements IAccordService, Shutdownable { TopologyRange remote = fetchTopologies(highestKnown + 1); - if (remote != null) - remote.forEach(configService::reportTopology, highestKnown + 1, Integer.MAX_VALUE); + if (remote != null) // TODO (required): if remote.min > highestKnown + 1, should we decide if we need to truncate our local topologies? Probably not until startup has finished. + remote.forEach(configService::reportTopology, remote.min, Integer.MAX_VALUE); } catch (InterruptedException e) { @@ -693,7 +674,7 @@ public class AccordService implements IAccordService, Shutdownable @Override public @Nonnull TxnResult coordinate(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, @Nonnull Dispatcher.RequestTime requestTime, long minHlc) throws RequestExecutionException { - return coordinateAsync(minEpoch, txn, consistencyLevel, requestTime, minHlc).awaitAndGet(); + return coordinateAsync(minEpoch, minHlc, txn, consistencyLevel, requestTime).awaitAndGet(); } @Override @@ -709,9 +690,9 @@ public class AccordService implements IAccordService, Shutdownable } @Override - public @Nonnull IAccordResult coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, @Nonnull Dispatcher.RequestTime requestTime, long minHlc) + public @Nonnull IAccordResult coordinateAsync(long minEpoch, long minHlc, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, @Nonnull Dispatcher.RequestTime requestTime) { - TxnId txnId = node.nextTxnId(minHlc >= 0 ? minHlc : 0, txn.kind(), txn.keys().domain(), cardinality(txn.keys())); + TxnId txnId = node.nextTxnId(minEpoch, minHlc, txn); long timeout = txnId.isWrite() ? DatabaseDescriptor.getWriteRpcTimeout(NANOSECONDS) : DatabaseDescriptor.getReadRpcTimeout(NANOSECONDS); ClientRequestBookkeeping bookkeeping = txn.isWrite() ? accordWriteBookkeeping : accordReadBookkeeping; bookkeeping.metrics.keySize.update(txn.keys().size()); diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index 0ad5f0c6dd..c6e754be8e 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -87,9 +87,9 @@ public interface IAccordService @Nonnull default IAccordResult coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime) { - return coordinateAsync(minEpoch, txn, consistencyLevel, requestTime, IAccordService.NO_HLC); + return coordinateAsync(minEpoch, IAccordService.NO_HLC, txn, consistencyLevel, requestTime); } - @Nonnull IAccordResult coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime, long minHlc); + @Nonnull IAccordResult coordinateAsync(long minEpoch, long minHlc, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime); @Nonnull default TxnResult coordinate(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime) { return coordinate(minEpoch, txn, consistencyLevel, requestTime, NO_HLC); @@ -244,7 +244,7 @@ public interface IAccordService } @Override - public @Nonnull IAccordResult coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime, long minHlc) + public @Nonnull IAccordResult coordinateAsync(long minEpoch, long minHlc, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime) { throw new UnsupportedOperationException("No accord transaction should be executed when accord.enabled = false in cassandra.yaml"); } @@ -454,9 +454,9 @@ public interface IAccordService @Nonnull @Override - public IAccordResult coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime, long minHlc) + public IAccordResult coordinateAsync(long minEpoch, long minHlc, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime) { - return delegate.coordinateAsync(minEpoch, txn, consistencyLevel, requestTime, minHlc); + return delegate.coordinateAsync(minEpoch, minHlc, txn, consistencyLevel, requestTime); } @Override diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java index 90df4950b1..d69abfd96e 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropAdapter.java @@ -115,6 +115,7 @@ public class AccordInteropAdapter extends TxnAdapter { Update update = txn.update(); ConsistencyLevel consistencyLevel = update instanceof AccordUpdate ? ((AccordUpdate) update).cassandraCommitCL() : null; + // TODO (expected): do we care about consistency level on recovery? we aren't reporting to the client so we shouldn't. if (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ANY || writes.isEmpty()) return false; diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java index fc093344b0..0e8f681a48 100644 --- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java +++ b/src/java/org/apache/cassandra/tcm/log/LocalLog.java @@ -24,10 +24,10 @@ import java.util.Collections; import java.util.Comparator; import java.util.HashSet; import java.util.List; -import java.util.NoSuchElementException; +import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.concurrent.ConcurrentSkipListSet; +import java.util.concurrent.ConcurrentSkipListMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; @@ -240,7 +240,7 @@ public abstract class LocalLog implements Closeable * However, snapshots should be applied out of order, and snapshots with higher epoch should be applied before snapshots * with a lower epoch in cases when there are multiple snapshots present. */ - protected final ConcurrentSkipListSet pending = new ConcurrentSkipListSet<>((Entry e1, Entry e2) -> { + protected final ConcurrentSkipListMap pending = new ConcurrentSkipListMap<>((Entry e1, Entry e2) -> { if (e1.transform.kind() == Transformation.Kind.FORCE_SNAPSHOT && e2.transform.kind() == Transformation.Kind.FORCE_SNAPSHOT) return e2.epoch.compareTo(e1.epoch); @@ -335,7 +335,7 @@ public abstract class LocalLog implements Closeable public boolean hasGaps() { Epoch start = committed.get().epoch; - for (Entry entry : pending) + for (Entry entry : pending.keySet()) { if (!entry.epoch.isDirectlyAfter(start)) return true; @@ -347,14 +347,8 @@ public abstract class LocalLog implements Closeable public Optional highestPending() { - try - { - return Optional.of(pending.last().epoch); - } - catch (NoSuchElementException eag) - { - return Optional.empty(); - } + Map.Entry e = pending.lastEntry(); + return e == null ? Optional.empty() : Optional.of(e.getKey().epoch); } public LogState getLocalEntries(Epoch since) @@ -381,7 +375,7 @@ public abstract class LocalLog implements Closeable { if (logger.isDebugEnabled()) logger.debug("Appending entries to the pending buffer: {}", entries.stream().map(e -> e.epoch).collect(Collectors.toList())); - pending.addAll(entries); + entries.forEach(e -> pending.put(e, true)); } processPending(); } @@ -433,7 +427,7 @@ public abstract class LocalLog implements Closeable } logger.debug("Appending entry to the pending buffer: {}", entry.epoch); - pending.add(entry); + pending.put(entry, true); } public abstract ClusterMetadata awaitAtLeast(Epoch epoch) throws InterruptedException, TimeoutException; @@ -459,14 +453,8 @@ public abstract class LocalLog implements Closeable private Entry peek() { - try - { - return pending.first(); - } - catch (NoSuchElementException ignore) - { - return null; - } + Map.Entry e = pending.firstEntry(); + return e == null ? null : e.getKey(); } /** @@ -571,7 +559,7 @@ public abstract class LocalLog implements Closeable } else { - Entry tmp = pending.first(); + Entry tmp = pending.firstKey(); if (tmp.epoch.is(pendingEntry.epoch)) { logger.debug("Smallest entry is non-consecutive {} to {}", pendingEntry.epoch, prev.epoch); @@ -745,7 +733,7 @@ public abstract class LocalLog implements Closeable { throw new TimeoutException(String.format("Timed out waiting for follower to run at least once. " + "Pending is %s and current is now at epoch %s.", - pending.stream().map((re) -> re.epoch).collect(Collectors.toList()), + pending.keySet().stream().map((re) -> re.epoch).collect(Collectors.toList()), metadata().epoch)); } } diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index 948087e2f9..4e63ce352b 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -152,7 +152,7 @@ public class AccordDebugKeyspaceTest extends CQLTester String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'"); AccordService accord = accord(); DatabaseDescriptor.getAccord().fetch_txn = "1s"; - TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); + TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key); Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0); execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS"); @@ -215,7 +215,7 @@ public class AccordDebugKeyspaceTest extends CQLTester { String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'"); AccordService accord = accord(); - TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); + TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key); Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0); String keyStr = txn.keys().get(0).toUnseekable().toString(); AsyncChains.getBlocking(accord.node().coordinate(id, txn)); @@ -237,7 +237,7 @@ public class AccordDebugKeyspaceTest extends CQLTester { String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'"); AccordService accord = accord(); - TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); + TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key); String insertTxn = String.format("BEGIN TRANSACTION\n" + " LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" + " IF r IS NULL THEN\n " + @@ -273,7 +273,7 @@ public class AccordDebugKeyspaceTest extends CQLTester { String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'"); AccordService accord = accord(); - TxnId first = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); + TxnId first = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key); String insertTxn = String.format("BEGIN TRANSACTION\n" + " LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" + " IF r IS NULL THEN\n " + @@ -292,7 +292,7 @@ public class AccordDebugKeyspaceTest extends CQLTester filter.reset(); - TxnId second = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); + TxnId second = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key); accord.node().coordinate(second, createTxn(insertTxn, 0, 0, 0, 0, 0)); filter.commit.awaitThrowUncheckedOnInterrupt(); @@ -356,7 +356,7 @@ public class AccordDebugKeyspaceTest extends CQLTester KEYSPACE, tableName); AccordService accord = accord(); - TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key); + TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key); accord.node().coordinate(id, createTxn(insertTxn, 0, 0, 0)); filter.preAccept.awaitThrowUncheckedOnInterrupt(); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java index 535bf7e528..d01817e88e 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java @@ -22,7 +22,6 @@ import java.net.UnknownHostException; import java.nio.file.Files; import java.util.ArrayList; import java.util.Collections; -import java.util.Iterator; import java.util.List; import java.util.Optional; import java.util.UUID; @@ -205,10 +204,10 @@ public class AccordConfigurationServiceTest listener = new AbstractConfigurationServiceTest.TestListener(loaded, true); loaded.registerListener(listener); journal_.closeCurrentSegmentForTestingIfNonEmpty(); - Iterator iter = journal.replayTopologies(); + List list = journal.replayTopologies(); // Simulate journal replay - while (iter.hasNext()) - loaded.reportTopology(iter.next().global); + for (TopologyUpdate update : list) + loaded.reportTopology(update.global); loaded.start(); listener.assertTopologiesFor(1L, 2L, 3L);