You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by if...@apache.org on 2025/08/14 07:40:04 UTC
(cassandra) branch trunk updated: Improve / simplify topology reporting on startup
This is an automated email from the ASF dual-hosted git repository.
ifesdjeen 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 2de6078289 Improve / simplify topology reporting on startup
2de6078289 is described below
commit 2de6078289b888b4151d07596096e9d4abfa2319
Author: Alex Petrov <ol...@gmail.com>
AuthorDate: Wed Jul 2 14:12:37 2025 +0200
Improve / simplify topology reporting on startup
Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20822
---
modules/accord | 2 +-
.../statements/schema/DropKeyspaceStatement.java | 2 +-
.../cql3/statements/schema/DropTableStatement.java | 4 +-
.../db/streaming/CassandraStreamReceiver.java | 2 +-
.../apache/cassandra/schema/DistributedSchema.java | 7 +-
.../apache/cassandra/service/CassandraDaemon.java | 2 +-
src/java/org/apache/cassandra/service/Rebuild.java | 9 +-
.../apache/cassandra/service/StorageService.java | 14 +-
.../service/accord/AccordConfigurationService.java | 28 +--
.../cassandra/service/accord/AccordService.java | 188 ++++++++++-------
.../cassandra/service/accord/EndpointMapping.java | 7 +
.../cassandra/service/accord/IAccordService.java | 27 ++-
.../org/apache/cassandra/streaming/StreamPlan.java | 6 +-
src/java/org/apache/cassandra/tcm/Startup.java | 6 -
.../org/apache/cassandra/tcm/log/LocalLog.java | 9 +-
.../cassandra/tcm/sequences/BootstrapAndJoin.java | 7 +-
.../org/apache/cassandra/tcm/sequences/Move.java | 6 +-
.../transformations/PrepareDropAccordTable.java | 2 +-
.../cassandra/tcm/transformations/Register.java | 9 +
.../cassandra/utils/JVMStabilityInspector.java | 2 +-
.../cassandra/distributed/impl/Instance.java | 1 -
.../cassandra/distributed/shared/ClusterUtils.java | 2 +-
.../test/PaxosEpochSerializationTest.java | 70 +++++++
.../distributed/test/PaxosRepairTest.java | 33 ---
.../test/accord/AccordIncrementalRepairTest.java | 222 +++++++++++----------
.../distributed/test/accord/AccordTestBase.java | 20 +-
.../cassandra/fuzz/topology/JournalGCTest.java | 2 +-
.../cassandra/simulator/ClusterSimulation.java | 7 +
.../simulator/paxos/AccordSimulationRunner.java | 5 +
.../org/apache/cassandra/dht/BootStrapperTest.java | 81 ++++----
.../cassandra/index/accord/RouteIndexTest.java | 41 ++--
31 files changed, 471 insertions(+), 352 deletions(-)
diff --git a/modules/accord b/modules/accord
index b86550c154..a8916a18a0 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit b86550c154758e5dd223fa8101a2a83005b644a6
+Subproject commit a8916a18a012063f4da74aaccd39f0b828c99da0
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
index e7159cda8d..0ab0447f6b 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropKeyspaceStatement.java
@@ -65,7 +65,7 @@ public final class DropKeyspaceStatement extends AlterSchemaStatement
.collect(Collectors.joining(",")));
List<TableMetadata> accordTables = keyspace.tables.stream()
- .filter(TableMetadata::isAccordEnabled)
+ .filter(TableMetadata::requiresAccordSupport)
.collect(Collectors.toList());
if (!accordTables.isEmpty())
throw ire("Cannot drop keyspace '%s' as it contains accord tables. (%s)",
diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
index a008b45488..c9e6516c57 100644
--- a/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/schema/DropTableStatement.java
@@ -60,7 +60,7 @@ public final class DropTableStatement extends AlterSchemaStatement
? null
: keyspace.getTableOrViewNullable(tableName);
if (table == null // this can happen when ifExists=true... since its already been validated can skip
- || !table.isAccordEnabled())
+ || !table.requiresAccordSupport())
return super.commit(metadata);
// Multi-Step Operation
@@ -94,7 +94,7 @@ public final class DropTableStatement extends AlterSchemaStatement
if (table.isView())
throw ire("Cannot use DROP TABLE on a materialized view. Please use DROP MATERIALIZED VIEW instead.");
- if (table.isAccordEnabled() && table.params.pendingDrop)
+ if (table.requiresAccordSupport() && table.params.pendingDrop)
throw ire("Table '%s.%s' is already being dropped", keyspaceName, tableName);
Iterable<ViewMetadata> views = keyspace.views.forTable(table.id);
diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
index fc3e9044cf..9bf1595a7c 100644
--- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
+++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
@@ -218,11 +218,11 @@ public class CassandraStreamReceiver implements StreamReceiver
{
CassandraVersion minVersion = ClusterMetadata.current().directory.clusterMinVersion.cassandraVersion;
checkNotNull(minVersion, "Unable to determine minimum cluster version");
- IAccordService accordService = AccordService.instance();
if (session.streamOperation().requiresBarrierTransaction()
&& cfs.metadata().requiresAccordSupport()
&& CassandraVersion.CASSANDRA_5_0.compareTo(minVersion) >= 0)
{
+ IAccordService accordService = AccordService.instance();
Ranges accordRanges = AccordTopology.toAccordRanges(cfs.getTableId(), ranges);
long startedAtNanos = nanoTime();
long timeoutNanos = DatabaseDescriptor.getAccordRangeSyncPointTimeoutNanos();
diff --git a/src/java/org/apache/cassandra/schema/DistributedSchema.java b/src/java/org/apache/cassandra/schema/DistributedSchema.java
index b13cfc7db3..0eceb9b169 100644
--- a/src/java/org/apache/cassandra/schema/DistributedSchema.java
+++ b/src/java/org/apache/cassandra/schema/DistributedSchema.java
@@ -56,7 +56,7 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
{
public static final Serializer serializer = new Serializer();
- public static final DistributedSchema empty()
+ public static DistributedSchema empty()
{
return new DistributedSchema(Keyspaces.none(), Epoch.EMPTY);
}
@@ -146,6 +146,11 @@ public class DistributedSchema implements MetadataValue<DistributedSchema>
return ks == null ? null : ks.tables.getNullable(cf);
}
+ public boolean hasAccordKeyspaces()
+ {
+ return keyspaces.stream().anyMatch(ksm -> ksm.tables.stream().anyMatch(TableMetadata::requiresAccordSupport));
+ }
+
public static DistributedSchema fromSystemTables(Keyspaces keyspaces, Set<String> knownDatacenters)
{
if (!keyspaces.containsKeyspace(SchemaConstants.METADATA_KEYSPACE_NAME))
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index be8bd058b9..d8acdbc266 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -824,7 +824,7 @@ public class CassandraDaemon
{
// Bootstrap with same address is an edge-case here, since we rely on HIBERNATE to prevent writes
// toward the bootstrapping replacement, so there's no startup sequence involved.
- if (StorageService.instance.isReplacingSameAddress() && StorageService.instance.isSurveyMode())
+ if (StorageService.isReplacingSameAddress() && StorageService.instance.isSurveyMode())
return;
// This node has not joined the ring (i.e. it was started with -Dcassandra.join_ring=false)
diff --git a/src/java/org/apache/cassandra/service/Rebuild.java b/src/java/org/apache/cassandra/service/Rebuild.java
index b6e4d61d1b..c7d40f08bf 100644
--- a/src/java/org/apache/cassandra/service/Rebuild.java
+++ b/src/java/org/apache/cassandra/service/Rebuild.java
@@ -158,10 +158,11 @@ public class Rebuild
streamer.addKeyspaceToFetch(keyspace);
}
- StreamResultFuture resultFuture = streamer.fetchAsync();
- // wait for result
- Future<Void> accordReady = AccordService.instance().epochReady(metadata.epoch);
- Future<?> ready = FutureCombiner.allOf(resultFuture, accordReady);
+ StreamResultFuture streamResult = streamer.fetchAsync();
+
+ Future<?> accordReady = AccordService.instance().epochReadyFor(metadata);
+ Future<?> ready = FutureCombiner.allOf(streamResult, accordReady);
+
// wait for result
ready.get();
}
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 87ce92f8f7..2c692932c3 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -844,18 +844,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
Gossiper.waitToSettle();
- NodeId self;
- if (isReplacingSameAddress())
- {
- self = ClusterMetadata.current().myNodeId();
- if (self == null)
- throw new IllegalStateException("Tried to replace same address, but node does not seem to be registered");
- }
- else
- {
- self = Register.maybeRegister();
- }
-
+ NodeId self = Register.maybeRegister();
+ AccordService.startup(self);
RegistrationStatus.instance.onRegistration();
Startup.maybeExecuteStartupTransformation(self);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
index 9f3563e367..37f8d74594 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
@@ -57,7 +57,6 @@ import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
-import org.apache.cassandra.tcm.listeners.ChangeListener;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Simulate;
@@ -132,17 +131,6 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
}
}
- //TODO (required): should not be public
- public final ChangeListener listener = new MetadataChangeListener();
- private class MetadataChangeListener implements ChangeListener
- {
- @Override
- public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
- {
- maybeReportMetadata(next);
- }
- }
-
public AccordConfigurationService(Node.Id node, Agent agent, MessageDelivery messagingService, IFailureDetector failureDetector, ScheduledExecutorPlus scheduledTasks)
{
super(node, agent);
@@ -187,7 +175,6 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
{
if (isTerminated())
return;
- ClusterMetadataService.instance().log().removeListener(listener);
state = State.SHUTDOWN;
}
@@ -235,8 +222,11 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
void reportMetadataInternal(ClusterMetadata metadata)
{
- updateMapping(metadata);
Topology topology = AccordTopology.createAccordTopology(metadata);
+ if (topology.isEmpty() && isEmpty())
+ return;
+
+ updateMapping(metadata);
if (Invariants.isParanoid())
{
for (Node.Id node : topology.nodes())
@@ -313,6 +303,10 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
long epoch = metadata.epoch.getEpoch();
synchronized (epochs)
{
+ // Accord has never been enabled for this cluster.
+ if (epochs.isEmpty() && !metadata.schema.hasAccordKeyspaces())
+ return;
+
// On first boot, we have 2 options:
//
// - we can start listening to TCM _before_ we replay topologies
@@ -496,8 +490,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
if (epoch < minEpoch() || epochs.wasTruncated(epoch))
return;
- Topology topology = getTopologyForEpoch(epoch);
- syncPropagator.reportClosed(epoch, topology.nodes(), ranges);
+ syncPropagator.reportClosed(epoch, mapping.nodes(), ranges);
}
@VisibleForTesting
@@ -514,8 +507,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
checkStarted();
// TODO (expected): ensure we aren't fetching a truncated epoch; otherwise this should be non-null
- Topology topology = getTopologyForEpoch(epoch);
- syncPropagator.reportRetired(epoch, topology.nodes(), ranges);
+ syncPropagator.reportRetired(epoch, mapping.nodes(), ranges);
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index a630a39b60..0b87121d41 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -29,6 +29,7 @@ import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.Function;
import javax.annotation.Nonnull;
@@ -39,6 +40,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import org.apache.cassandra.utils.Clock;
+import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -102,7 +104,6 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
-import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.AccordSyncPropagator.Notification;
import org.apache.cassandra.service.accord.TimeOnlyRequestBookkeeping.LatencyRequestBookkeeping;
import org.apache.cassandra.service.accord.api.AccordAgent;
@@ -125,6 +126,7 @@ import org.apache.cassandra.service.consensus.migration.TableMigrationState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
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.ExecutorUtils;
@@ -160,7 +162,59 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class AccordService implements IAccordService, Shutdownable
{
+ public static class MetadataChangeListener implements ChangeListener.Async
+ {
+ // Listener is initialized before Accord is initialized
+ public static MetadataChangeListener instance = new MetadataChangeListener();
+
+ private MetadataChangeListener() {}
+
+ private final AtomicReference<ChangeListener> collector = new AtomicReference<>(new PreInitStateCollector());
+
+ @Override
+ public void notifyPreCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
+ {
+ collector.get().notifyPreCommit(prev, next, fromSnapshot);
+ }
+
+ @Override
+ public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
+ {
+ collector.get().notifyPostCommit(prev, next, fromSnapshot);
+ }
+
+ @VisibleForTesting
+ public void resetForTesting(ClusterMetadata metadata)
+ {
+ PreInitStateCollector stateCollector = new PreInitStateCollector();
+ stateCollector.items.add(metadata);
+ collector.set(stateCollector);
+ }
+
+ /**
+ * Collects TCM events from startup util full Accord initialization to avoid races with TCM and creating gaps between
+ * epochs restored from journal and reported by TCM.
+ **/
+
+ static class PreInitStateCollector implements ChangeListener
+ {
+ private final List<ClusterMetadata> items = new ArrayList<>(4);
+
+ @Override
+ public synchronized void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
+ {
+ items.add(next);
+ }
+
+ public synchronized List<ClusterMetadata> getItems()
+ {
+ return new ArrayList<>(items);
+ }
+ }
+ }
+
private static final Logger logger = LoggerFactory.getLogger(AccordService.class);
+ private static final Future<Void> EPOCH_READY = ImmediateFuture.success(null);
static
{
ProtocolModifiers.Toggles.setPermitLocalExecution(true);
@@ -227,34 +281,42 @@ public class AccordService implements IAccordService, Shutdownable
return instance().responseHandler();
}
- public synchronized static void startup(NodeId tcmId)
+ @VisibleForTesting
+ public synchronized static AccordService startup(NodeId tcmId)
{
if (!DatabaseDescriptor.getAccordTransactionsEnabled())
{
instance = NOOP_SERVICE;
- return;
+ return null;
}
if (instance != null)
- return;
+ return (AccordService) instance;
AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmId));
as.startup();
- if (StorageService.instance.isReplacingSameAddress())
- {
- // 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.configService().listener.notifyPostCommit(current, current, false);
- }
instance = as;
replayJournal(as);
+ as.finishInitialization();
+
+ as.configService.start();
+ as.configService.unsafeMarkTruncated();
+ as.fastPathCoordinator.start();
+
+ ClusterMetadataService.instance().log().addListener(as.fastPathCoordinator);
+ as.node.durability().shards().reconfigure(Ints.checkedCast(getAccordShardDurabilityTargetSplits()),
+ Ints.checkedCast(getAccordShardDurabilityMaxSplits()),
+ Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS);
+ as.node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS);
+ as.state = State.STARTED;
// Only enable durability scheduling _after_ we have fully replayed journal
as.configService.registerListener(as.node.durability());
as.node.durability().start();
+
+ WatermarkCollector.fetchAndReportWatermarksAsync(as.configService);
+ return as;
}
@VisibleForTesting
@@ -355,12 +417,6 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public synchronized void startup()
- {
- unsafeStartupWithOverrides(null);
- }
-
- @VisibleForTesting
- public synchronized void unsafeStartupWithOverrides(@Nullable Journal.TopologyUpdate overrideNullTopologyUpdate)
{
if (state != State.INIT)
return;
@@ -370,7 +426,6 @@ public class AccordService implements IAccordService, Shutdownable
ClusterMetadata metadata = ClusterMetadata.current();
configService.updateMapping(metadata);
- long highestKnown = -1;
List<ImmutableTopoloyImage> images = new ArrayList<>();
// Collect locally known topologies
@@ -387,66 +442,55 @@ public class AccordService implements IAccordService, Shutdownable
prev = next;
}
- if (prev == null)
- prev = overrideNullTopologyUpdate;
-
// Instantiate latest topology from the log, if known
if (prev != null)
{
node.commandStores().initializeTopologyUnsafe(prev);
- highestKnown = prev.global.epoch();
}
+ // Replay local epochs
+ for (ImmutableTopoloyImage image : images)
+ configService.reportTopology(image.global);
+ }
+
+ /**
+ * Startup is broken up in two phases: local and distributed startup. During local startup, we replay up to
+ * the latest epoch known to the node prior to restart. After that, we replay journal itself, and only after
+ * that we finish initializaiton and replay the rest of epochs.
+ */
+ @VisibleForTesting
+ public void finishInitialization()
+ {
+ configService.updateMapping(ClusterMetadata.current());
+ long highestKnown = -1;
+ if (configService.currentTopology() != null)
+ highestKnown = configService.currentEpoch();
try
{
TopologyRange remote = fetchTopologies(highestKnown + 1);
- // Replay local epochs
- for (ImmutableTopoloyImage image : images)
- configService.reportTopology(image.global);
-
if (remote != null)
remote.forEach(configService::reportTopology, highestKnown + 1, Integer.MAX_VALUE);
// Subscribe to TCM events
- ClusterMetadataService.instance().log().addListener(configService.listener);
-
- // We report current topology _after_ subscribing to TCM events since in a single-node cluster there
- // will be no notification about the current epoch, since it's already reported. And in a multi-node cluster
- // we do not want a race between addinga listener and reporting an epoch.
- if (remote == null && images.isEmpty())
- configService.reportTopology(AccordTopology.createAccordTopology(metadata));
-
- WatermarkCollector.fetchAndReportWatermarksAsync(configService());
- configService.unsafeMarkTruncated();
-
- int attempt = 0;
- int waitSeconds = 5;
- long deadine = Clock.Global.nanoTime() + SECONDS.toNanos(60);
- while (true)
+ ChangeListener prevListener = MetadataChangeListener.instance.collector.getAndSet(new ChangeListener()
{
- Epoch await = Epoch.max(Epoch.create(configService.currentEpoch()), metadata.epoch);
- try
- {
- epochReady(await).get(waitSeconds, SECONDS);
- break;
- }
- catch (TimeoutException e)
+ @Override
+ public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
{
- logger.warn("Epoch {} is not ready after waiting for {} seconds", metadata.epoch, (++attempt) * waitSeconds);
- // In case there are any gaps, fetch unknown topologies.
- metadata = ClusterMetadata.current();
- highestKnown = configService.currentEpoch();
- if (metadata.epoch.getEpoch() > highestKnown)
- {
- remote = fetchTopologies(highestKnown + 1);
- if (remote != null)
- remote.forEach(configService::reportTopology, highestKnown + 1, Integer.MAX_VALUE);
- }
+ if (state != State.SHUTDOWN)
+ configService.maybeReportMetadata(next);
}
+ });
- if (Clock.Global.nanoTime() > deadine)
- throw new IllegalStateException(String.format("Could not initialize epoch %s. Config service state:\n%s", await, configService.getDebugStr()));
+ Invariants.require((prevListener instanceof MetadataChangeListener.PreInitStateCollector),
+ "Listener should have been initialized with Accord pre-init state collector, but was " + prevListener.getClass());
+
+ MetadataChangeListener.PreInitStateCollector preinit = (MetadataChangeListener.PreInitStateCollector) prevListener;
+ for (ClusterMetadata item : preinit.getItems())
+ {
+ if (item.epoch.getEpoch() > minEpoch())
+ configService.maybeReportMetadata(item);
}
}
catch (InterruptedException e)
@@ -458,17 +502,7 @@ public class AccordService implements IAccordService, Shutdownable
{
throw new RuntimeException(e);
}
-
- configService.start();
- fastPathCoordinator.start();
- ClusterMetadataService.instance().log().addListener(fastPathCoordinator);
- node.durability().shards().reconfigure(Ints.checkedCast(getAccordShardDurabilityTargetSplits()),
- Ints.checkedCast(getAccordShardDurabilityMaxSplits()),
- Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS);
- node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS);
- state = State.STARTED;
}
-
/**
* Queries peers to discover min epoch, and then fetches all topologies between min and current epochs
*/
@@ -938,9 +972,8 @@ public class AccordService implements IAccordService, Shutdownable
return cmdTxnState.build();
}
- @Nullable
@Override
- public Long minEpoch()
+ public long minEpoch()
{
return node.topology().minEpoch();
}
@@ -973,6 +1006,15 @@ public class AccordService implements IAccordService, Shutdownable
return promise;
}
+ @Override
+ public Future<Void> epochReadyFor(ClusterMetadata metadata)
+ {
+ if (!metadata.schema.hasAccordKeyspaces())
+ return EPOCH_READY;
+
+ return epochReady(metadata.epoch);
+ }
+
@Override
public void receive(Message<Notification> message)
{
diff --git a/src/java/org/apache/cassandra/service/accord/EndpointMapping.java b/src/java/org/apache/cassandra/service/accord/EndpointMapping.java
index 78ec8c4589..1d5a21995a 100644
--- a/src/java/org/apache/cassandra/service/accord/EndpointMapping.java
+++ b/src/java/org/apache/cassandra/service/accord/EndpointMapping.java
@@ -18,6 +18,8 @@
package org.apache.cassandra.service.accord;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import com.google.common.collect.BiMap;
@@ -55,6 +57,11 @@ class EndpointMapping implements AccordEndpointMapper
return mapping.containsKey(id);
}
+ public List<Node.Id> nodes()
+ {
+ return new ArrayList<>(mapping.keySet());
+ }
+
public Map<Node.Id, Long> removedNodes()
{
return removedNodes;
diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java
index 40c0e57a4c..1c5d624aab 100644
--- a/src/java/org/apache/cassandra/service/accord/IAccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java
@@ -28,6 +28,7 @@ import java.util.function.BiConsumer;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import org.apache.cassandra.tcm.ClusterMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -126,6 +127,8 @@ public interface IAccordService
*/
Future<Void> epochReady(Epoch epoch);
+ Future<Void> epochReadyFor(ClusterMetadata epoch);
+
void receive(Message<AccordSyncPropagator.Notification> message);
class AccordCompactionInfo
@@ -173,8 +176,8 @@ public interface IAccordService
Id nodeId();
List<CommandStoreTxnBlockedGraph> debugTxnBlockedGraph(TxnId txnId);
- @Nullable
- Long minEpoch();
+
+ long minEpoch();
void awaitDone(TableId id, long epoch);
@@ -302,6 +305,12 @@ public interface IAccordService
return BOOTSTRAP_SUCCESS;
}
+ @Override
+ public Future<Void> epochReadyFor(ClusterMetadata epoch)
+ {
+ return BOOTSTRAP_SUCCESS;
+ }
+
@Override
public void receive(Message<AccordSyncPropagator.Notification> message) {}
@@ -329,11 +338,10 @@ public interface IAccordService
return Collections.emptyList();
}
- @Nullable
@Override
- public Long minEpoch()
+ public long minEpoch()
{
- return null;
+ return -1;
}
@Override
@@ -498,6 +506,12 @@ public interface IAccordService
return delegate.epochReady(epoch);
}
+ @Override
+ public Future<Void> epochReadyFor(ClusterMetadata epoch)
+ {
+ return delegate.epochReadyFor(epoch);
+ }
+
@Override
public void receive(Message<Notification> message)
{
@@ -528,9 +542,8 @@ public interface IAccordService
return delegate.debugTxnBlockedGraph(txnId);
}
- @Nullable
@Override
- public Long minEpoch()
+ public long minEpoch()
{
return delegate.minEpoch();
}
diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java
index 93b864d79b..9c39e5dbd2 100644
--- a/src/java/org/apache/cassandra/streaming/StreamPlan.java
+++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java
@@ -238,7 +238,7 @@ public class StreamPlan
public static String[] nonAccordTablesForKeyspace(KeyspaceMetadata ksm)
{
String[] result = ksm.tables.stream()
- .filter(tbl -> !tbl.isAccordEnabled())
+ .filter(tbl -> !tbl.requiresAccordSupport())
.map(tbl -> tbl.name)
.toArray(String[]::new);
@@ -247,11 +247,11 @@ public class StreamPlan
public static boolean hasNonAccordTables(KeyspaceMetadata ksm)
{
- return ksm.tables.stream().anyMatch(tbl -> !tbl.isAccordEnabled());
+ return ksm.tables.stream().anyMatch(tbl -> !tbl.requiresAccordSupport());
}
public static boolean hasAccordTables(KeyspaceMetadata ksm)
{
- return ksm.tables.stream().anyMatch(TableMetadata::isAccordEnabled);
+ return ksm.tables.stream().anyMatch(TableMetadata::requiresAccordSupport);
}
}
diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java
index 64ccbf37e8..bb660b9845 100644
--- a/src/java/org/apache/cassandra/tcm/Startup.java
+++ b/src/java/org/apache/cassandra/tcm/Startup.java
@@ -53,7 +53,6 @@ import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
-import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.tcm.log.LocalLog;
import org.apache.cassandra.tcm.log.LogStorage;
import org.apache.cassandra.tcm.log.SystemKeyspaceStorage;
@@ -76,7 +75,6 @@ import static org.apache.cassandra.tcm.compatibility.GossipHelper.emptyWithSchem
import static org.apache.cassandra.tcm.compatibility.GossipHelper.fromEndpointStates;
import static org.apache.cassandra.tcm.membership.NodeState.JOINED;
import static org.apache.cassandra.tcm.membership.NodeState.LEFT;
-import static org.apache.cassandra.tcm.membership.NodeState.REGISTERED;
import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
/**
@@ -423,9 +421,6 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
metadata = ClusterMetadata.current();
NodeState startingstate = metadata.directory.peerState(self);
- if (startingstate != REGISTERED && startingstate != LEFT)
- AccordService.startup(self);
-
switch (startingstate)
{
case REGISTERED:
@@ -437,7 +432,6 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
// When Accord starts up it needs to check for any historic epochs that it needs to know about (in order
// to handle pending transactions), in order to know what nodes to check with it needs to know what the
// settled placement is (so it knows what peers to reach out to).
- AccordService.startup(self);
InProgressSequences.finishInProgressSequences(self, true); // potentially finish the MSO committed above
metadata = ClusterMetadata.current();
diff --git a/src/java/org/apache/cassandra/tcm/log/LocalLog.java b/src/java/org/apache/cassandra/tcm/log/LocalLog.java
index 425f4bdc24..fc093344b0 100644
--- a/src/java/org/apache/cassandra/tcm/log/LocalLog.java
+++ b/src/java/org/apache/cassandra/tcm/log/LocalLog.java
@@ -38,6 +38,7 @@ import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import org.apache.cassandra.service.accord.AccordService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -507,7 +508,7 @@ public abstract class LocalLog implements Closeable
}
catch (Throwable t)
{
- logger.error(String.format("Caught an exception while processing entry %s. This can mean that this node is configured differently from CMS.", prev), t);
+ logger.error("Caught an exception while processing entry {}. This can mean that this node is configured differently from CMS.", prev, t);
throw new StopProcessingException(t);
}
@@ -564,8 +565,8 @@ public abstract class LocalLog implements Closeable
}
else if (!pendingEntry.epoch.isAfter(metadata().epoch))
{
- logger.debug(String.format("An already appended entry %s discovered in the pending buffer, ignoring. Max consecutive: %s",
- pendingEntry.epoch, prev.epoch));
+ logger.debug("An already appended entry {} discovered in the pending buffer, ignoring. Max consecutive: {}",
+ pendingEntry.epoch, prev.epoch);
pending.remove(pendingEntry);
}
else
@@ -937,6 +938,8 @@ public abstract class LocalLog implements Closeable
addListener(new MetadataSnapshotListener());
addListener(new ClientNotificationListener());
addListener(new UpgradeMigrationListener());
+ if (DatabaseDescriptor.getAccord().enabled)
+ addListener(AccordService.MetadataChangeListener.instance);
}
private LogListener snapshotListener()
diff --git a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
index 0e32f18c25..341eff845b 100644
--- a/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
@@ -25,7 +25,6 @@ import java.util.Set;
import java.util.stream.StreamSupport;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -363,14 +362,16 @@ public class BootstrapAndJoin extends MultiStepOperation<Epoch>
StorageService.instance.repairPaxosForTopologyChange("bootstrap");
Future<StreamState> bootstrapStream = StorageService.instance.startBootstrap(metadata, beingReplaced, movements, strictMovements);
- Future<Void> accordReady = AccordService.instance().epochReady(metadata.epoch);
- Future<?> ready = FutureCombiner.allOf(Lists.newArrayList(bootstrapStream, accordReady));
+ Future<?> accordReady = AccordService.instance().epochReadyFor(metadata);
+ Future<?> ready = FutureCombiner.allOf(bootstrapStream, accordReady);
+
try
{
if (bootstrapTimeoutMillis > 0)
ready.get(bootstrapTimeoutMillis, MILLISECONDS);
else
ready.get();
+
StorageService.instance.markViewsAsBuilt();
StorageService.instance.clearOngoingBootstrap();
logger.info("Bootstrap completed for tokens {}", tokens);
diff --git a/src/java/org/apache/cassandra/tcm/sequences/Move.java b/src/java/org/apache/cassandra/tcm/sequences/Move.java
index 450889694e..b54b796749 100644
--- a/src/java/org/apache/cassandra/tcm/sequences/Move.java
+++ b/src/java/org/apache/cassandra/tcm/sequences/Move.java
@@ -257,8 +257,10 @@ public class Move extends MultiStepOperation<Epoch>
}
StreamResultFuture streamResult = streamPlan.execute();
- Future<Void> accordReady = AccordService.instance().epochReady(metadata.epoch);
- FutureCombiner.allOf(streamResult, accordReady).get();
+
+ Future<?> accordReady = AccordService.instance().epochReadyFor(metadata);
+ Future<?> ready = FutureCombiner.allOf(streamResult, accordReady);
+ ready.get();
StorageService.instance.repairPaxosForTopologyChange("move");
}
catch (InterruptedException e)
diff --git a/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java b/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java
index 2c960a0bd3..80c33816c7 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/PrepareDropAccordTable.java
@@ -58,7 +58,7 @@ public class PrepareDropAccordTable implements Transformation
TableMetadata metadata = prev.schema.getKeyspaces().getTableOrViewNullable(tableRef.id);
if (metadata == null)
return new Rejected(ExceptionCode.INVALID, "Table " + tableRef + " is not known");
- if (!metadata.isAccordEnabled())
+ if (!metadata.requiresAccordSupport())
return new Rejected(ExceptionCode.INVALID, "Table " + metadata + " is not an Accord table and should be dropped normally");
if (metadata.params.pendingDrop)
return new Rejected(ExceptionCode.INVALID, "Table " + metadata + " is in the process of being dropped");
diff --git a/src/java/org/apache/cassandra/tcm/transformations/Register.java b/src/java/org/apache/cassandra/tcm/transformations/Register.java
index 7cf45fab8b..c60244bab7 100644
--- a/src/java/org/apache/cassandra/tcm/transformations/Register.java
+++ b/src/java/org/apache/cassandra/tcm/transformations/Register.java
@@ -46,6 +46,7 @@ import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.exceptions.ExceptionCode.INVALID;
+import static org.apache.cassandra.service.StorageService.isReplacingSameAddress;
public class Register implements Transformation
{
@@ -86,6 +87,14 @@ public class Register implements Transformation
public static NodeId maybeRegister()
{
+ if (isReplacingSameAddress())
+ {
+ NodeId self = ClusterMetadata.current().myNodeId();
+ if (self == null)
+ throw new IllegalStateException("Tried to replace same address, but node does not seem to be registered");
+
+ return self;
+ }
return register(false);
}
diff --git a/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java b/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java
index e77a231da2..0ca992b642 100644
--- a/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java
+++ b/src/java/org/apache/cassandra/utils/JVMStabilityInspector.java
@@ -27,13 +27,13 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
+import ch.qos.logback.classic.LoggerContext;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import ch.qos.logback.classic.LoggerContext;
import net.nicoulaj.compilecommand.annotations.Exclude;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.Config;
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index c8e24ef092..70743f7b06 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@ -844,7 +844,6 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
JVMStabilityInspector.replaceKiller(new InstanceKiller(Instance.this::shutdown));
StorageService.instance.registerDaemon(CassandraDaemon.getInstanceForTesting());
-
if (config.has(GOSSIP))
{
try
diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
index a9fe78f642..51a551a19a 100644
--- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
+++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
@@ -699,7 +699,7 @@ public class ClusterUtils
return maxEpoch(cluster, up.toIntArray());
}
- public static Epoch maxEpoch(ICluster<IInvokableInstance> cluster, int[] nodes)
+ public static Epoch maxEpoch(ICluster<IInvokableInstance> cluster, int... nodes)
{
Epoch max = null;
for (int id : nodes)
diff --git a/test/distributed/org/apache/cassandra/distributed/test/PaxosEpochSerializationTest.java b/test/distributed/org/apache/cassandra/distributed/test/PaxosEpochSerializationTest.java
new file mode 100644
index 0000000000..ab23628fa3
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/PaxosEpochSerializationTest.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.transformations.CustomTransformation;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.cassandra.net.Verb.PAXOS_COMMIT_REQ;
+
+public class PaxosEpochSerializationTest extends TestBaseImpl
+{
+ private static final Logger logger = LoggerFactory.getLogger(PaxosEpochSerializationTest.class);
+ private static final String TABLE = "tbl";
+
+ @Test
+ public void epochBadDeserializationTest() throws Throwable
+ {
+ try (Cluster cluster = init(Cluster.build(3).withConfig(cfg -> cfg.with(Feature.NETWORK)
+ .with(Feature.GOSSIP))
+ .withoutVNodes().start()))
+ {
+ cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk text primary key, v int)");
+ cluster.get(1).runOnInstance(() -> {
+ // just execute transformations to get epoch bumped enough for the bug to occur
+ for (int i = 0; i < 75; i++)
+ ClusterMetadataService.instance().commit(CustomTransformation.make("x"+i));
+ });
+ // and bump the epoch in the tablemetadata:
+ cluster.schemaChange("ALTER TABLE " + KEYSPACE + '.' + TABLE + " WITH comment='abc'");
+
+ cluster.verbs(PAXOS_COMMIT_REQ).drop();
+ try
+ {
+ cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, v) VALUES ('xyzxyzxyzxyzxyzxyzxyzxyz', 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
+ Assert.fail("expected write timeout");
+ }
+ catch (RuntimeException e)
+ {
+ // exception expected
+ }
+
+ cluster.filters().reset();
+ cluster.get(1).shutdown().get();
+ cluster.get(1).startup();
+ }
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/PaxosRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/PaxosRepairTest.java
index febffe5290..d18f3a6f06 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/PaxosRepairTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/PaxosRepairTest.java
@@ -91,7 +91,6 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.NodeVersion;
-import org.apache.cassandra.tcm.transformations.CustomTransformation;
import org.apache.cassandra.tcm.transformations.ForceSnapshot;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.CassandraVersion;
@@ -295,38 +294,6 @@ public class PaxosRepairTest extends TestBaseImpl
}
}
- @Test
- public void epochBadDeserializationTest() throws Throwable
- {
- try (Cluster cluster = init(Cluster.build(3).withConfig(WITH_NETWORK).withoutVNodes().start()))
- {
- cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk text primary key, v int)");
- cluster.get(1).runOnInstance(() -> {
- // just execute transformations to get epoch bumped enough for the bug to occur
- for (int i = 0; i < 75; i++)
- ClusterMetadataService.instance().commit(CustomTransformation.make("x"+i));
- });
- // and bump the epoch in the tablemetadata:
- cluster.schemaChange("ALTER TABLE " + KEYSPACE + '.' + TABLE + " WITH comment='abc'");
-
- cluster.verbs(PAXOS_COMMIT_REQ).drop();
- try
- {
- cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, v) VALUES ('xyzxyzxyzxyzxyzxyzxyzxyz', 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
- Assert.fail("expected write timeout");
- }
- catch (RuntimeException e)
- {
- // exception expected
- }
-
- cluster.filters().reset();
- cluster.get(1).shutdown().get();
- cluster.get(1).startup();
- }
- }
-
-
@Test
public void topologyChangePaxosTest() throws Throwable
{
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
index a5aae94516..7f9fef5a6a 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIncrementalRepairTest.java
@@ -19,22 +19,10 @@
package org.apache.cassandra.distributed.test.accord;
import java.util.Collection;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;
-
-import com.google.common.collect.Iterables;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import accord.local.Node;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
@@ -49,14 +37,18 @@ import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
+import com.google.common.collect.Iterables;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.harry.checker.ModelChecker;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
@@ -68,14 +60,20 @@ import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Clock;
-import org.apache.cassandra.utils.concurrent.Future;
-import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static java.lang.String.format;
+import static org.apache.cassandra.distributed.test.accord.AccordTestBase.executeWithRetry;
-public class AccordIncrementalRepairTest extends AccordTestBase
+public class AccordIncrementalRepairTest extends TestBaseImpl
{
private static final Logger logger = LoggerFactory.getLogger(AccordIncrementalRepairTest.class);
@@ -122,7 +120,6 @@ public class AccordIncrementalRepairTest extends AccordTestBase
return AccordService.instance();
}
- @Override
protected Logger logger()
{
return logger;
@@ -131,25 +128,41 @@ public class AccordIncrementalRepairTest extends AccordTestBase
@BeforeClass
public static void setupClass() throws Throwable
{
- setupCluster(opt -> opt.withConfig(conf -> conf.with(Feature.NETWORK, Feature.GOSSIP)
- .set("accord.recover_txn", "1s")
- .set("accord.retry_syncpoint", "1s*attempts")
- .set("accord.retry_durability", "1s*attempts")
- .set("accord.shard_durability_target_splits", 4)
- ), 3);
- for (IInvokableInstance instance : SHARED_CLUSTER)
+
+
+ }
+
+ private static Cluster createCluster() throws Throwable
+ {
+ Cluster cluster = AccordTestBase.createCluster(3, opt -> opt.withConfig(conf -> conf.with(Feature.NETWORK, Feature.GOSSIP)
+ .set("accord.recover_txn", "1s")
+ .set("accord.retry_syncpoint", "1s*attempts")
+ .set("accord.retry_durability", "1s*attempts")
+ .set("accord.shard_durability_target_splits", 4)));
+ for (IInvokableInstance instance : cluster)
instance.runOnInstance(() -> AccordService.unsafeSetNewAccordService(new BarrierRecordingService(AccordService.instance())));
-// setupCluster(opt -> opt, 3);
+
+ return cluster;
}
- @After
- public void tearDown()
+ private static void withCluster(ModelChecker.ThrowingConsumer<Cluster> run) throws Throwable
{
- for (IInvokableInstance instance : SHARED_CLUSTER)
- instance.runOnInstance(() -> AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().start()));
- SHARED_CLUSTER.filters().reset();
+ try (Cluster cluster = createCluster())
+ {
+ try
+ {
+ run.accept(cluster);
+ }
+ catch (Throwable t)
+ {
+ cluster.filters().reset();
+ for (IInvokableInstance instance : cluster)
+ instance.runOnInstance(() -> AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().start()));
+ }
+ }
}
+
private static void await(IInvokableInstance instance, IIsolatedExecutor.SerializableCallable<Boolean> check, long duration, TimeUnit unit)
{
instance.runOnInstance(() -> {
@@ -184,27 +197,6 @@ public class AccordIncrementalRepairTest extends AccordTestBase
await(instance, () -> !FailureDetector.instance.isAlive(endpoint), 1, TimeUnit.MINUTES);
}
- private static <V> V getUninterruptibly(Future<V> future, long timeout, TimeUnit units)
- {
- try
- {
- return future.get(timeout, units);
- }
- catch (InterruptedException e)
- {
- throw new UncheckedInterruptedException(e);
- }
- catch (ExecutionException | TimeoutException e)
- {
- throw new RuntimeException(e);
- }
- }
-
- private static <V> V getUninterruptibly(Future<V> future)
- {
- return getUninterruptibly(future, 1, TimeUnit.MINUTES);
- }
-
private static TxnId awaitLocalApplyOnKey(TableMetadata metadata, int k)
{
return awaitLocalApplyOnKey(new TokenKey(metadata.id, metadata.partitioner.decorateKey(ByteBufferUtil.bytes(k)).getToken()));
@@ -254,25 +246,31 @@ public class AccordIncrementalRepairTest extends AccordTestBase
@Test
public void txnRepairTest() throws Throwable
{
- SHARED_CLUSTER.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, accordTableName));
+ withCluster(this::txnRepairTest);
+ }
+
+ public void txnRepairTest(Cluster cluster) throws Throwable
+ {
final String keyspace = KEYSPACE;
- final String table = accordTableName;
+ final String table = "accord_table";
+ final String qualifiedTable = String.format("%s.%s", keyspace, table);
+ cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", keyspace, table));
- SHARED_CLUSTER.filters().allVerbs().to(3).drop();
- awaitEndpointDown(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
+ cluster.filters().allVerbs().to(3).drop();
+ awaitEndpointDown(cluster.get(1), cluster.get(3));
- executeWithRetry(SHARED_CLUSTER, format("BEGIN TRANSACTION\n" +
- "INSERT INTO %s (k, v) VALUES (1, 1);\n" +
- "COMMIT TRANSACTION", qualifiedAccordTableName));
+ executeWithRetry(cluster, format("BEGIN TRANSACTION\n" +
+ "INSERT INTO %s (k, v) VALUES (1, 1);\n" +
+ "COMMIT TRANSACTION", qualifiedTable));
- SHARED_CLUSTER.get(1, 2).forEach(instance -> instance.runOnInstance(() -> {
+ cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> {
TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table);
awaitLocalApplyOnKey(metadata, 1);
}));
- SHARED_CLUSTER.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset()));
+ cluster.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset()));
- SHARED_CLUSTER.get(1, 2).forEach(instance -> {
+ cluster.get(1, 2).forEach(instance -> {
instance.runOnInstance(() -> {
ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
@@ -283,23 +281,23 @@ public class AccordIncrementalRepairTest extends AccordTestBase
});
});
});
- SHARED_CLUSTER.get(3).runOnInstance(() -> {
+ cluster.get(3).runOnInstance(() -> {
ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
Assert.assertTrue(cfs.getLiveSSTables().isEmpty());
});
// heal partition and wait for node 1 to see node 3 again
- for (IInvokableInstance instance : SHARED_CLUSTER)
+ for (IInvokableInstance instance : cluster)
instance.runOnInstance(() -> {
AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().stop());
Assert.assertFalse(barrierRecordingService().executedBarriers);
});
- SHARED_CLUSTER.filters().reset();
- awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
- nodetool(SHARED_CLUSTER.get(1), "repair", KEYSPACE);
+ cluster.filters().reset();
+ awaitEndpointUp(cluster.get(1), cluster.get(3));
+ nodetool(cluster.get(1), "repair", keyspace);
- SHARED_CLUSTER.get(1).runOnInstance(() -> {
+ cluster.get(1).runOnInstance(() -> {
Assert.assertTrue(barrierRecordingService().executedBarriers);
ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
Assert.assertFalse(cfs.getLiveSSTables().isEmpty());
@@ -309,17 +307,17 @@ public class AccordIncrementalRepairTest extends AccordTestBase
});
}
- private void testSingleNodeWrite(TransactionalMode mode)
+ private void testSingleNodeWrite(Cluster cluster, TransactionalMode mode)
{
- SHARED_CLUSTER.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='%s';", KEYSPACE, accordTableName, mode));
final String keyspace = KEYSPACE;
- final String table = accordTableName;
+ final String table = "accord_table";
+ cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='%s';", keyspace, table, mode));
- SHARED_CLUSTER.get(3).runOnInstance(() -> {
+ cluster.get(3).runOnInstance(() -> {
QueryProcessor.executeInternal(String.format("INSERT INTO %s.%s (k, v) VALUES (1, 2);", keyspace, table));
});
- SHARED_CLUSTER.get(3).runOnInstance(() -> {
+ cluster.get(3).runOnInstance(() -> {
UntypedResultSet result = QueryProcessor.executeInternal(format("SELECT * FROM %s.%s WHERE k=1", keyspace, table));
Assert.assertFalse(result.isEmpty());
UntypedResultSet.Row row = Iterables.getOnlyElement(result);
@@ -336,7 +334,7 @@ public class AccordIncrementalRepairTest extends AccordTestBase
Assert.assertFalse(sstable.isPendingRepair());
});
});
- SHARED_CLUSTER.get(1, 2).forEach(instance -> instance.runOnInstance(() -> {
+ cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> {
UntypedResultSet result = QueryProcessor.executeInternal(format("SELECT * FROM %s.%s WHERE k=1", keyspace, table));
Assert.assertTrue(result.isEmpty());
@@ -344,12 +342,12 @@ public class AccordIncrementalRepairTest extends AccordTestBase
cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
Assert.assertTrue(cfs.getLiveSSTables().isEmpty());
}));
- SHARED_CLUSTER.forEach(instance -> instance.runOnInstance(() -> {
+ cluster.forEach(instance -> instance.runOnInstance(() -> {
barrierRecordingService().reset();
}));
- nodetool(SHARED_CLUSTER.get(1), "repair", KEYSPACE);
- SHARED_CLUSTER.get(1).runOnInstance(() -> {
+ nodetool(cluster.get(1), "repair", KEYSPACE);
+ cluster.get(1).runOnInstance(() -> {
Assert.assertTrue(barrierRecordingService().executedBarriers);
ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
Assert.assertFalse(cfs.getLiveSSTables().isEmpty());
@@ -369,74 +367,90 @@ public class AccordIncrementalRepairTest extends AccordTestBase
* a failed write at txn mode unsafe should be made visible by repair
*/
@Test
- public void unsafeRepairTest()
+ public void unsafeRepairTest() throws Throwable
{
- testSingleNodeWrite(TransactionalMode.test_unsafe);
+ withCluster(cluster -> {
+ testSingleNodeWrite(cluster, TransactionalMode.test_unsafe);
+ });
}
/**
* Repair should repair (fully replicate _some_ state) any divergent state between replicas
*/
@Test
- public void fullRepairTest()
+ public void fullRepairTest() throws Throwable
{
- testSingleNodeWrite(TransactionalMode.full);
+ withCluster(cluster -> {
+ testSingleNodeWrite(cluster, TransactionalMode.full);
+ });
}
@Test
- public void onlyAccordTest()
+ public void onlyAccordTest() throws Throwable
+ {
+ withCluster(this::onlyAccordTest);
+ }
+
+ public void onlyAccordTest(Cluster cluster)
{
- SHARED_CLUSTER.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, accordTableName));
final String keyspace = KEYSPACE;
- final String table = accordTableName;
+ final String table = "accord_table";
+ final String qualifiedTable = String.format("%s.%s", keyspace, table);
+ cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, table));
- executeWithRetry(SHARED_CLUSTER, format("BEGIN TRANSACTION\n" +
+ executeWithRetry(cluster, format("BEGIN TRANSACTION\n" +
"INSERT INTO %s (k, v) VALUES (1, 1);\n" +
- "COMMIT TRANSACTION", qualifiedAccordTableName));
+ "COMMIT TRANSACTION", qualifiedTable));
- SHARED_CLUSTER.get(1, 2).forEach(instance -> instance.runOnInstance(() -> {
+ cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> {
TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table);
awaitLocalApplyOnKey(metadata, 1);
}));
- SHARED_CLUSTER.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset()));
+ cluster.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset()));
- SHARED_CLUSTER.filters().reset();
- awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
- nodetool(SHARED_CLUSTER.get(1), "repair", "--accord-only", KEYSPACE);
+ cluster.filters().reset();
+ awaitEndpointUp(cluster.get(1), cluster.get(3));
+ nodetool(cluster.get(1), "repair", "--accord-only", KEYSPACE);
- SHARED_CLUSTER.get(1).runOnInstance(() -> {
+ cluster.get(1).runOnInstance(() -> {
Assert.assertTrue(barrierRecordingService().executedBarriers);
});
}
@Test
- public void onlyAccordWithForceTest()
+ public void onlyAccordWithForceTest() throws Throwable
+ {
+ withCluster(this::onlyAccordWithForceTest);
+ }
+
+ public void onlyAccordWithForceTest(Cluster cluster)
{
- SHARED_CLUSTER.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, accordTableName));
final String keyspace = KEYSPACE;
- final String table = accordTableName;
+ final String table = "accord_table";
+ final String qualifiedTable = String.format("%s.%s", keyspace, table);
+ cluster.schemaChange(format("CREATE TABLE %s.%s (k int primary key, v int) WITH transactional_mode='full' AND fast_path={'size':2};", KEYSPACE, table));
- SHARED_CLUSTER.filters().allVerbs().to(3).drop();
- awaitEndpointDown(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
- awaitEndpointDown(SHARED_CLUSTER.get(2), SHARED_CLUSTER.get(3));
+ cluster.filters().allVerbs().to(3).drop();
+ awaitEndpointDown(cluster.get(1), cluster.get(3));
+ awaitEndpointDown(cluster.get(2), cluster.get(3));
- executeWithRetry(SHARED_CLUSTER, format("BEGIN TRANSACTION\n" +
+ executeWithRetry(cluster, format("BEGIN TRANSACTION\n" +
"INSERT INTO %s (k, v) VALUES (1, 1);\n" +
- "COMMIT TRANSACTION", qualifiedAccordTableName));
+ "COMMIT TRANSACTION", qualifiedTable));
- SHARED_CLUSTER.get(1, 2).forEach(instance -> instance.runOnInstance(() -> {
+ cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> {
TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table);
awaitLocalApplyOnKey(metadata, 1);
}));
- SHARED_CLUSTER.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset()));
+ cluster.forEach(instance -> instance.runOnInstance(() -> barrierRecordingService().reset()));
- SHARED_CLUSTER.filters().reset();
- awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
- nodetool(SHARED_CLUSTER.get(1), "repair", "--force", "--accord-only", KEYSPACE);
+ cluster.filters().reset();
+ awaitEndpointUp(cluster.get(1), cluster.get(3));
+ nodetool(cluster.get(1), "repair", "--force", "--accord-only", KEYSPACE);
- SHARED_CLUSTER.get(1).runOnInstance(() -> {
+ cluster.get(1).runOnInstance(() -> {
Assert.assertTrue(barrierRecordingService().executedBarriers);
});
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
index 2236f27cd1..6334ff6957 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
@@ -111,6 +111,7 @@ import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.db.SystemKeyspace.CONSENSUS_MIGRATION_STATE;
import static org.apache.cassandra.db.SystemKeyspace.PAXOS;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.maxEpoch;
import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME;
import static org.junit.Assert.assertArrayEquals;
@@ -163,17 +164,22 @@ public abstract class AccordTestBase extends TestBaseImpl
@After
public void tearDown() throws Exception
{
+ SHARED_CLUSTER.filters().reset();
for (IInvokableInstance instance : SHARED_CLUSTER)
instance.runOnInstance(() -> AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().start()));
truncateSystemTables();
- ClusterUtils.waitForCMSToQuiesce(SHARED_CLUSTER, 1);
- SHARED_CLUSTER.forEach(() -> Util.spinUntilTrue(() -> ClusterMetadata.current().epoch.getEpoch() ==
- ((AccordService) AccordService.instance()).configService().currentEpoch() &&
- AccordService.instance().topology().current().epoch() ==
- ((AccordService) AccordService.instance()).configService().currentEpoch(),
- 60));
+ ClusterUtils.waitForCMSToQuiesce(SHARED_CLUSTER, maxEpoch(SHARED_CLUSTER, 1), true);
+ SHARED_CLUSTER.forEach(() -> Util.spinUntilTrue(() -> {
+ ClusterMetadata metadata = ClusterMetadata.current();
+ if (!metadata.schema.hasAccordKeyspaces())
+ return true;
+
+ AccordService accord = ((AccordService) AccordService.instance());
+ return metadata.epoch.getEpoch() == accord.configService().currentEpoch() &&
+ metadata.epoch.getEpoch() == accord.topology().current().epoch();
+ }, 60));
}
protected static void assertRowSerial(Cluster cluster, String query, int k, int c, int v, int s)
@@ -370,7 +376,7 @@ public abstract class AccordTestBase extends TestBaseImpl
instance.runOnInstance(runnable);
}
- private static Cluster createCluster(int nodes, Function<Builder, Builder> options) throws IOException
+ public static Cluster createCluster(int nodes, Function<Builder, Builder> options) throws IOException
{
// need to up the timeout else tests get flaky
// disable vnode for now, but should enable before trunk
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java
index 4fc4d59534..064d050207 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java
@@ -127,7 +127,7 @@ public class JournalGCTest extends FuzzTestBase
}, maximumId);
int after =-1;
- int maxCycles = 3;
+ int maxCycles = 10;
for (int i = 0; i < maxCycles; i++)
{
cluster.get(1).acceptOnInstance((ks, tbl) -> {
diff --git a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
index fb9e3f41ae..ada4e51a34 100644
--- a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
+++ b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
@@ -40,6 +40,7 @@ import com.google.common.util.concurrent.FutureCallback;
import org.apache.cassandra.auth.PasswordSaltSupplier;
import org.apache.cassandra.concurrent.ExecutorFactory;
+import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.Constants;
@@ -96,6 +97,7 @@ import org.apache.cassandra.simulator.utils.KindOfSequence;
import org.apache.cassandra.simulator.utils.LongRange;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.Closeable;
+import org.apache.cassandra.utils.StorageCompatibilityMode;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.memory.BufferPool;
@@ -119,6 +121,11 @@ import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
@SuppressWarnings("RedundantCast")
public class ClusterSimulation<S extends Simulation> implements AutoCloseable
{
+ static
+ {
+ CassandraRelevantProperties.TEST_STORAGE_COMPATIBILITY_MODE.setEnum(StorageCompatibilityMode.NONE);
+ }
+
public static final Class<?>[] SHARE = new Class[]
{
AsyncFunction.class,
diff --git a/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java b/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java
index 5fed781d18..162aa45b2f 100644
--- a/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java
+++ b/test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java
@@ -40,6 +40,11 @@ import picocli.CommandLine.Command;
AccordSimulationRunner.Reconcile.class})
public class AccordSimulationRunner extends SimulationRunner
{
+ static
+ {
+ CassandraRelevantProperties.TEST_STORAGE_COMPATIBILITY_MODE.setString(StorageCompatibilityMode.NONE.toString());
+ }
+
@BeforeClass
public static void beforeAll()
{
diff --git a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
index c139acbe3c..4cb8b2fd0e 100644
--- a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
+++ b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.dht;
import java.net.UnknownHostException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Random;
@@ -31,6 +32,7 @@ import com.google.common.base.Predicates;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import org.junit.AfterClass;
+import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -47,7 +49,6 @@ import org.apache.cassandra.gms.IFailureDetector;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.schema.Schema;
-import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamOperation;
import org.apache.cassandra.tcm.ClusterMetadata;
@@ -107,16 +108,16 @@ public class BootStrapperTest
@Test
public void testSourceTargetComputation() throws UnknownHostException
{
- final int[] clusterSizes = new int[] { 1, 3, 5, 10, 100 };
+ final int[] clusterSizes = new int[]{1, 3, 5, 10, 100};
+ List<String> keyspaces = new ArrayList<>();
for (String keyspaceName : Schema.instance.getNonLocalStrategyKeyspaces().names())
{
- if (keyspaceName.equals(SchemaConstants.METADATA_KEYSPACE_NAME))
- continue;
- int replicationFactor = Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor().allReplicas;
- for (int clusterSize : clusterSizes)
- if (clusterSize >= replicationFactor)
- testSourceTargetComputation(keyspaceName, clusterSize, replicationFactor);
+ if (keyspaceName.startsWith("BootStrapperTest"))
+ keyspaces.add(keyspaceName);
}
+
+ for (int clusterSize : clusterSizes)
+ testSourceTargetComputation(keyspaces, clusterSize);
}
@Test
@@ -155,39 +156,47 @@ public class BootStrapperTest
}
}
- private RangeStreamer testSourceTargetComputation(String keyspaceName, int numOldNodes, int replicationFactor) throws UnknownHostException
+ private void testSourceTargetComputation(List<String> keyspaces, int clusterSize) throws UnknownHostException
{
ServerTestUtils.resetCMS();
- generateFakeEndpoints(numOldNodes);
+ generateFakeEndpoints(clusterSize);
ClusterMetadata metadata = ClusterMetadata.current();
- assertEquals(numOldNodes, metadata.tokenMap.tokens().size());
+ assertEquals(clusterSize, metadata.tokenMap.tokens().size());
+
+ InetAddressAndPort myEndpoint = InetAddressAndPort.getByName("127.0.0.1");
RangeStreamer s = getRangeStreamer();
+ for (String keyspace : keyspaces)
+ {
+ assertNotNull(Keyspace.open(keyspace));
+ int replicationFactor = Keyspace.open(keyspace).getReplicationStrategy().getReplicationFactor().allReplicas;
+ if (clusterSize < replicationFactor)
+ continue;
- assertNotNull(Keyspace.open(keyspaceName));
- s.addKeyspaceToFetch(keyspaceName);
- Multimap<InetAddressAndPort, RangeStreamer.FetchReplica> toFetch = s.toFetch().get(keyspaceName);
-
- // Pre-TCM this test would always run with RangeStreamer::useStrictSourcesForRanges returning false because
- // the RangeStreamer instance was constructed with a null Collection<Token>, relying on pending ranges being
- // calculated in the test code, and then cloning TokenMetadata with pending tokens added to pass to
- // calculateRangesToFetchWithPreferredEndpoints. Post-TCM, we calculate both relaxed and strict movements
- // together and TokenMetadata is no more, so the equivalent operation now ends up using strict movements. For
- // that reason, when RF includes transient replicas, toFetch will include both a transient and full source,
- // hence we dedupe the ranges here.
- Set<Range<Token>> fetchRanges = toFetch.values()
- .stream()
- .map(fr -> fr.remote.range())
- .collect(Collectors.toSet());
- // Post CEP-21 wrapping ranges are also unwrapped at this point, so account for that when setting expectation
- int expectedRangeCount = replicationFactor + (includesWraparound(fetchRanges) ? 1 : 0);
- assertEquals(expectedRangeCount, fetchRanges.size());
-
- // there isn't any point in testing the size of these collections for any specific size. When a random partitioner
- // is used, they will vary.
- assert toFetch.values().size() > 0;
-
- assert toFetch.keys().stream().noneMatch(InetAddressAndPort.getByName("127.0.0.1")::equals);
- return s;
+
+ s.addKeyspaceToFetch(keyspace);
+ Multimap<InetAddressAndPort, RangeStreamer.FetchReplica> toFetch = s.toFetch().get(keyspace);
+
+ // Pre-TCM this test would always run with RangeStreamer::useStrictSourcesForRanges returning false because
+ // the RangeStreamer instance was constructed with a null Collection<Token>, relying on pending ranges being
+ // calculated in the test code, and then cloning TokenMetadata with pending tokens added to pass to
+ // calculateRangesToFetchWithPreferredEndpoints. Post-TCM, we calculate both relaxed and strict movements
+ // together and TokenMetadata is no more, so the equivalent operation now ends up using strict movements. For
+ // that reason, when RF includes transient replicas, toFetch will include both a transient and full source,
+ // hence we dedupe the ranges here.
+ Set<Range<Token>> fetchRanges = toFetch.values()
+ .stream()
+ .map(fr -> fr.remote.range())
+ .collect(Collectors.toSet());
+ // Post CEP-21 wrapping ranges are also unwrapped at this point, so account for that when setting expectation
+ int expectedRangeCount = replicationFactor + (includesWraparound(fetchRanges) ? 1 : 0);
+ assertEquals(expectedRangeCount, fetchRanges.size());
+
+ // there isn't any point in testing the size of these collections for any specific size. When a random partitioner
+ // is used, they will vary.
+ Assert.assertTrue(toFetch.values().size() > 0);
+
+ Assert.assertTrue(toFetch.keys().stream().noneMatch(myEndpoint::equals));
+ }
}
private RangeStreamer getRangeStreamer() throws UnknownHostException
diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
index 77f379571f..7b32878e71 100644
--- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
@@ -29,18 +29,9 @@ import java.util.Set;
import java.util.TreeSet;
import java.util.function.Supplier;
import java.util.stream.Collectors;
-
import javax.annotation.Nullable;
-
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Sets;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
import accord.api.Journal;
import accord.api.RoutingKey;
-import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.DurableBefore;
@@ -58,13 +49,13 @@ import accord.primitives.Route;
import accord.primitives.SaveStatus;
import accord.primitives.Txn;
import accord.primitives.TxnId;
-import accord.topology.Shard;
-import accord.topology.Topology;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.Property.Command;
import accord.utils.Property.UnitCommand;
import accord.utils.RandomSource;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
import org.agrona.collections.Int2ObjectHashMap;
import org.agrona.collections.Long2ObjectHashMap;
import org.agrona.collections.ObjectHashSet;
@@ -83,12 +74,10 @@ import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableId;
-import org.apache.cassandra.service.accord.AccordCommandStore;
import org.apache.cassandra.service.accord.AccordJournal;
import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.AccordTestUtils;
-import org.apache.cassandra.service.accord.AccordTopology;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfo;
import org.apache.cassandra.service.accord.TokenRange;
@@ -103,12 +92,14 @@ import org.apache.cassandra.utils.RTree;
import org.apache.cassandra.utils.RangeTree;
import org.apache.cassandra.utils.concurrent.CountDownLatch;
import org.assertj.core.api.Assertions;
-import org.mockito.Mockito;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
import static accord.local.RedundantStatus.SomeStatus.NONE;
import static accord.utils.Property.commands;
import static accord.utils.Property.stateful;
-import static accord.utils.SortedArrays.SortedArrayList.ofSorted;
import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner;
import static org.apache.cassandra.index.accord.AccordIndexUtil.normalize;
import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
@@ -512,7 +503,6 @@ public class RouteIndexTest extends CQLTester
storeRangesForEpochs.put(i, new RangesForEpoch(1, Ranges.of(TokenRange.fullRange(tableId, getPartitioner()))));
accordService = startAccord();
- accordService.configService().listener.notifyPostCommit(null, ClusterMetadata.current(), false);
accordService.epochReady(ClusterMetadata.current().epoch).awaitUninterruptibly();
minDecidedIdNull = rs.nextFloat();
@@ -525,19 +515,11 @@ public class RouteIndexTest extends CQLTester
AccordService startAccord()
{
- NodeId tcmNodeId = ClusterMetadata.current().myNodeId();
- AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmNodeId));
- Topology topology = new Topology(1, Shard.create(TokenRange.fullRange(tableId, getPartitioner()), ofSorted(new Node.Id(1)), ofSorted(new Node.Id(1))));
- as.unsafeStartupWithOverrides(new Journal.TopologyUpdate(storeRangesForEpochs, topology));
- for (CommandStore commandStore : as.node().commandStores().all())
- ((AccordCommandStore)commandStore).unsafeUpsertRedundantBefore(emptyRedundantBefore);
- // the reason for the mocking is to speed up compaction. Collecting the info from the stores has been slow and its always empty in this test... so stub it out to speed up the test
- AccordService mock = Mockito.spy(as);
- Mockito.doReturn(emptyCompactionInfo(tableId, emptyRedundantBefore, storeRangesForEpochs)).when(mock).getCompactionInfo();
- AccordService.unsafeSetNewAccordService(mock);
-
- AccordService.replayJournal(as);
- return as;
+ ClusterMetadata metadata = ClusterMetadata.current();
+ AccordService.MetadataChangeListener.instance.resetForTesting(metadata);
+ NodeId tcmNodeId = metadata.myNodeId();
+ AccordService.unsafeSetNewAccordService(null);
+ return AccordService.startup(tcmNodeId);
}
TxnId nextTxnId(Domain domain)
@@ -626,6 +608,7 @@ public class RouteIndexTest extends CQLTester
private void restartAccord()
{
+ accordService.journal().closeCurrentSegmentForTestingIfNonEmpty();
accordService.shutdown();
accordService = startAccord();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org