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

Fix AccordMigrationTest, preclude possible races in topology propagation 

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20316
This commit is contained in:
Alex Petrov 2025-02-07 19:04:05 +01:00 committed by David Capwell
parent 64c4aa88f5
commit 67e21ec4ba
13 changed files with 101 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 1542da5d8acf08b8226227c91d4f81c64ed9762c
Subproject commit a4a8255bebb6804b8f55289f2fbc846a5253e66c

64
src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
View File

@ -22,7 +22,9 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
@ -59,7 +61,6 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Simulate;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.service.accord.AccordTopology.tcmIdToAccord;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
@ -448,21 +449,60 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
getOrCreateEpochState(epoch - 1).acknowledged().addCallback(() -> reportMetadata(metadata));
}
private final Map<Long, Future<Topology>> pendingTopologies = new ConcurrentHashMap<>();
@Override
protected void fetchTopologyInternal(long epoch)
public void fetchTopologyForEpoch(long epoch)
{
long minEpoch = currentEpoch() + 1;
// Find and fetch all epochs in-between
for (long i = minEpoch; i <= epoch; ++i)
fetchTopologyInternal(i);
}
private void fetchTopologyInternal(long epoch)
{
pendingTopologies.computeIfAbsent(epoch, (epoch_) -> {
AsyncPromise<Topology> future = new AsyncPromise<>();
fetchTopologyAsync(epoch_,
(topology, throwable) -> {
if (topology != null)
future.setSuccess(topology);
else
{
Invariants.require(future == pendingTopologies.remove(epoch_));
future.setFailure(Invariants.nonNull(throwable));
fetchTopologyForEpoch(epoch_);
}
});
return future;
});
}
private void fetchTopologyAsync(long epoch, BiConsumer<Topology, ? super Throwable> onResult)
{
// It's not safe for this to block on CMS so for now pick a thread pool to handle it
Stage.ACCORD_MIGRATION.execute(() -> {
if (ClusterMetadata.current().epoch.getEpoch() < epoch)
ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(epoch));
try
{
if (ClusterMetadata.current().epoch.getEpoch() < epoch)
ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(epoch));
}
catch (Throwable t)
{
onResult.accept(null, t);
return;
}
// In most cases, after fetching log from CMS, we will be caught up to the required epoch.
// This TCM will also notify Accord via reportMetadata, so we do not need to fetch topologies.
// If metadata has reported has skipped one or more epochs, and is _ahead_ of the requested epoch,
// we need to fetch topologies from peers to fill in the gap.
ClusterMetadata metadata = ClusterMetadata.current();
if (metadata.epoch.getEpoch() == epoch)
{
onResult.accept(AccordTopology.createAccordTopology(metadata), null);
return;
}
try
{
@ -475,18 +515,18 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
Topology topology = FetchTopology.fetch(SharedContext.Global.instance, peers, epoch).get();
Invariants.require(topology.epoch() == epoch);
reportTopology(topology);
}
catch (InterruptedException e)
{
if (currentEpoch() >= epoch)
return;
Thread.currentThread().interrupt();
throw new UncheckedInterruptedException(e);
onResult.accept(topology, null);
}
catch (Throwable e)
{
if (currentEpoch() >= epoch)
{
onResult.accept(getTopologyForEpoch(epoch), null);
return;
}
if (e instanceof InterruptedException)
Thread.currentThread().interrupt();
onResult.accept(null, e);
throw new RuntimeException(e.getCause());
}
});

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

@ -440,7 +440,6 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
@SuppressWarnings("unchecked") @Override
public void replay(CommandStores commandStores)
{
journal.closeCurrentSegmentForTestingIfNonEmpty();
try (CloseableIterator<Journal.KeyRefs<JournalKey>> iter = journalTable.keyIterator())
{
while (iter.hasNext())

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

@ -261,8 +261,8 @@ public class AccordService implements IAccordService, Shutdownable
// when replacing another node but using the same ip the hostId will also match, this causes no TCM transactions
// to be committed...
// In order to bootup correctly, need to pull in the current epoch
ClusterMetadata current = ClusterMetadata.current();
as.configurationService().listener.notifyPostCommit(current, current, false);
ClusterMetadata metadata = ClusterMetadata.current();
as.configurationService().listener.notifyPostCommit(metadata, metadata, false);
}
instance = as;
@ -381,6 +381,7 @@ public class AccordService implements IAccordService, Shutdownable
{
Journal.TopologyUpdate update = iter.next();
local.add(update.global);
Invariants.require(lastSeen == null || update.global.epoch() > lastSeen.global.epoch());
lastSeen = update;
}

17
src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java
View File

@ -72,17 +72,20 @@ public class AccordVerbHandler<T extends Request> implements IVerbHandler<T>
request.process(node, fromNodeId, message.header);
else
{
// TODO (required): review this claim. Downstream from `withEpoch`, we do call fetch log, albeit from _CMS_, since we
// do not know the peer there.
// withEpoch does not reliably ensure that TCM is up to date, if Accord has the topology it won't
// wait for TCM to come up to date, so do it here in the verb handler
if (!cmUpToDate)
{
ClusterMetadataService.instance().fetchLogFromPeerOrCMSAsync(cm, message.from(), Epoch.create(waitForEpoch)).addCallback((success, failure) ->
node.withEpoch(waitForEpoch, (ignored, withEpochFailure) -> {
if (withEpochFailure != null)
throw new RuntimeException("Timed out waiting for epoch when processing message from " + fromNodeId + " to " + node + " message " + message, withEpochFailure);
request.process(node, fromNodeId, message.header);
})
);
ClusterMetadataService.instance().fetchLogFromPeerOrCMSAsync(cm, message.from(), Epoch.create(waitForEpoch))
.addCallback((success, failure) -> {
node.withEpoch(waitForEpoch, (ignored, withEpochFailure) -> {
if (withEpochFailure != null)
throw new RuntimeException("Timed out waiting for epoch when processing message from " + fromNodeId + " to " + node + " message " + message, withEpochFailure);
request.process(node, fromNodeId, message.header);
});
});
}
else
{

16
test/distributed/org/apache/cassandra/distributed/test/accord/AccordMigrationWriteRaceTestBase.java
View File

@ -275,6 +275,22 @@ public abstract class AccordMigrationWriteRaceTestBase extends AccordTestBase
ConsensusKeyMigrationState.reset();
});
truncateSystemTables();
ClusterUtils.waitForCMSToQuiesce(SHARED_CLUSTER, 1);
SHARED_CLUSTER.forEach(() -> Util.spinUntilTrue(() -> ClusterMetadata.current().epoch.getEpoch() ==
((AccordService) AccordService.instance()).configurationService().currentEpoch() &&
AccordService.instance().topology().current().epoch() ==
((AccordService) AccordService.instance()).configurationService().currentEpoch(),
60));
SHARED_CLUSTER.forEach(() -> {
try
{
AccordService.instance().epochReady(ClusterMetadata.current().epoch).get(30, TimeUnit.SECONDS);
}
catch (Throwable e)
{
throw new RuntimeException(e);
}
});
}
private ListenableFuture<Void> alterTableTransactionalModeAsync(TransactionalMode mode)

2
test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
View File

@ -371,6 +371,8 @@ public abstract class AccordTestBase extends TestBaseImpl
.set("write_request_timeout", "10s")
.set("transaction_timeout", "15s")
.set("native_transport_timeout", "30s")
.set("cms_await_timeout", "1s")
.set("cms_default_max_retries", 10_000)
.set("accord.ephemeral_read_enabled", "false")
.set("accord.shard_durability_target_splits", "1")
.set("accord.shard_durability_cycle", "60s")

3
test/distributed/org/apache/cassandra/distributed/test/accord/MigrationFromAccordWriteRaceTest.java
View File

@ -18,9 +18,6 @@
package org.apache.cassandra.distributed.test.accord;
import org.junit.Ignore;
@Ignore
public class MigrationFromAccordWriteRaceTest extends AccordMigrationWriteRaceTestBase
{
protected boolean migratingAwayFromAccord()

1
test/distributed/org/apache/cassandra/fuzz/ring/ConsistentBootstrapTest.java
View File

@ -153,6 +153,7 @@ public class ConsistentBootstrapTest extends FuzzTestBase
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
.withConfig((config) -> config.with(Feature.NETWORK, Feature.GOSSIP)
.set("write_request_timeout", "10s")
.set("accord.enabled", false)
.set("metadata_snapshot_frequency", 5))
.start())
{

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

@ -32,6 +32,7 @@ import accord.burn.SimulationException;
import accord.impl.TopologyFactory;
import accord.impl.basic.Cluster;
import accord.impl.basic.RandomDelayQueue;
import accord.local.CommandStores;
import accord.local.Node;
import accord.utils.DefaultRandom;
import accord.utils.RandomSource;
@ -55,7 +56,6 @@ import org.apache.cassandra.tools.FieldUtil;
import static accord.impl.PrefixedIntHashKey.ranges;
public class AccordJournalBurnTest extends BurnTestBase
{
private static final Logger logger = LoggerFactory.getLogger(AccordJournalBurnTest.class);
@ -163,7 +163,14 @@ public class AccordJournalBurnTest extends BurnTestBase
{
return false;
}
}, new AccordAgent(), directory, cfs);
}, new AccordAgent(), directory, cfs)
{
public void replay(CommandStores commandStores)
{
closeCurrentSegmentForTestingIfNonEmpty();
super.replay(commandStores);
}
};
journal.start(null);
journal.unsafeSetStarted();

3
test/simulator/main/org/apache/cassandra/simulator/cluster/ClusterActions.java
View File

@ -61,6 +61,7 @@ import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.transformations.UnsafeJoin;
import static org.apache.cassandra.simulator.Action.Modifiers.NO_TIMEOUTS;
import static org.apache.cassandra.simulator.Action.Modifiers.RELIABLE;
import static org.apache.cassandra.simulator.Debug.EventType.CLUSTER;
import static org.apache.cassandra.simulator.cluster.ClusterActions.TopologyChange.JOIN;
import static org.apache.cassandra.simulator.cluster.ClusterActions.TopologyChange.LEAVE;
@ -216,7 +217,7 @@ public class ClusterActions extends SimulatedSystems
return StrictAction.of("Initialise Cluster", () -> {
List<Action> actions = new ArrayList<>();
cluster.stream().forEach(i -> actions.add(invoke("Startup " + i.broadcastAddress(), NO_TIMEOUTS, NO_TIMEOUTS,
cluster.stream().forEach(i -> actions.add(invoke("Startup " + i.broadcastAddress(), RELIABLE, RELIABLE,
new InterceptedRunnableExecution((InterceptingExecutor) i.executor(), i::startup))));
List<InetSocketAddress> endpoints = cluster.stream().map(IInstance::broadcastAddress).collect(Collectors.toList());

4
test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosSimulation.java
View File

@ -39,6 +39,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.coordinate.CoordinationFailed;
import accord.coordinate.Invalidated;
import accord.coordinate.Preempted;
import org.apache.cassandra.concurrent.ExecutorFactory;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.distributed.Cluster;
@ -90,6 +92,8 @@ public abstract class PaxosSimulation implements Simulation, ClusterActionListen
protected Class<? extends Throwable>[] expectedExceptionsAccord()
{
return (Class<? extends Throwable>[]) new Class<?>[] { RequestExecutionException.class,
Invalidated.class,
Preempted.class,
CancellationException.class,
CoordinationFailed.class,
ClosedChannelException.class,

2
test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java
View File

@ -429,7 +429,7 @@ public class AccordSyncPropagatorTest
}
@Override
protected void fetchTopologyInternal(long epoch)
public void fetchTopologyForEpoch(long epoch)
{
// TODO
}