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

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
This commit is contained in:
Benedict Elliott Smith 2025-09-03 14:51:52 +01:00
parent 5c7d8d8982
commit b91731b5a3
9 changed files with 58 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 48061a521cde7a085fe7a5503023c8ef8b687a89
Subproject commit c3a62d77ba332f4a01220709c040af86dd3acf6a

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

@ -370,9 +370,10 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
}
@Override
public CloseableIterator<TopologyUpdate> replayTopologies()
public List<TopologyUpdate> replayTopologies()
{
return new CloseableIterator<>()
List<TopologyUpdate> images = new ArrayList<>();
try (CloseableIterator<TopologyUpdate> iter = new CloseableIterator<>()
{
final CloseableIterator<Journal.KeyRefs<JournalKey>> 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<JournalKey> 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

2
src/java/org/apache/cassandra/service/accord/AccordJournalTable.java
View File

@ -567,7 +567,7 @@ public class AccordJournalTable<K extends JournalKey, V> 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()

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

@ -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<TopologyUpdate> images = new ArrayList<>();
TopologyUpdate prev = null;
// Collect locally known topologies
try (CloseableIterator<TopologyUpdate> 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<TopologyUpdate> 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<TxnResult> coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, @Nonnull Dispatcher.RequestTime requestTime, long minHlc)
public @Nonnull IAccordResult<TxnResult> 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());

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

@ -87,9 +87,9 @@ public interface IAccordService
@Nonnull
default IAccordResult<TxnResult> 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<TxnResult> coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime, long minHlc);
@Nonnull IAccordResult<TxnResult> 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<TxnResult> coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime, long minHlc)
public @Nonnull IAccordResult<TxnResult> 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<TxnResult> coordinateAsync(long minEpoch, @Nonnull Txn txn, @Nonnull ConsistencyLevel consistencyLevel, RequestTime requestTime, long minHlc)
public IAccordResult<TxnResult> 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

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

@ -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;

36
src/java/org/apache/cassandra/tcm/log/LocalLog.java
View File

@ -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<Entry> pending = new ConcurrentSkipListSet<>((Entry e1, Entry e2) -> {
protected final ConcurrentSkipListMap<Entry, Boolean> 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<Epoch> highestPending()
{
try
{
return Optional.of(pending.last().epoch);
}
catch (NoSuchElementException eag)
{
return Optional.empty();
}
Map.Entry<Entry, Boolean> 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<Entry, Boolean> 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));
}
}

12
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -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();

7
test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java
View File

@ -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<TopologyUpdate> iter = journal.replayTopologies();
List<TopologyUpdate> 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);