Improve / simplify topology reporting on startup

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20822

modules/accord

@@ -1 +1 @@
-Subproject commit b86550c154758e5dd223fa8101a2a83005b644a6
+Subproject commit a8916a18a012063f4da74aaccd39f0b828c99da0

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)",

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

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

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))

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)

src/java/org/apache/cassandra/service/Rebuild.java

@@ -158,10 +158,11 @@ public class Rebuild
                 streamer.addKeyspaceToFetch(keyspace);
             }
 
-            StreamResultFuture resultFuture = streamer.fetchAsync();
+            StreamResultFuture streamResult = streamer.fetchAsync();
-            // wait for result
+
-            Future<Void> accordReady = AccordService.instance().epochReady(metadata.epoch);
+            Future<?> accordReady = AccordService.instance().epochReadyFor(metadata);
-            Future<?> ready = FutureCombiner.allOf(resultFuture, accordReady);
+            Future<?> ready = FutureCombiner.allOf(streamResult, accordReady);
+
             // wait for result
             ready.get();
         }

src/java/org/apache/cassandra/service/StorageService.java

@@ -844,18 +844,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         }
         Gossiper.waitToSettle();
 
-        NodeId self;
+        NodeId self = Register.maybeRegister();
-        if (isReplacingSameAddress())
+        AccordService.startup(self);
-        {
-            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();
-        }
-
         RegistrationStatus.instance.onRegistration();
         Startup.maybeExecuteStartupTransformation(self);
 

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, mapping.nodes(), ranges);
-        syncPropagator.reportClosed(epoch, topology.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, mapping.nodes(), ranges);
-        syncPropagator.reportRetired(epoch, topology.nodes(), ranges);
     }
 
     @Override

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);
+            ChangeListener prevListener = MetadataChangeListener.instance.collector.getAndSet(new ChangeListener()
-
-            // 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)
             {
-                Epoch await = Epoch.max(Epoch.create(configService.currentEpoch()), metadata.epoch);
+                @Override
-                try
+                public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
                 {
-                    epochReady(await).get(waitSeconds, SECONDS);
+                    if (state != State.SHUTDOWN)
-                    break;
+                        configService.maybeReportMetadata(next);
-                }
-                catch (TimeoutException e)
-                {
-                    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 (Clock.Global.nanoTime() > deadine)
+            Invariants.require((prevListener instanceof MetadataChangeListener.PreInitStateCollector),
-                    throw new IllegalStateException(String.format("Could not initialize epoch %s. Config service state:\n%s", await, configService.getDebugStr()));
+                               "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)
     {

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;

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

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);
     }
 }

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

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",
+                logger.debug("An already appended entry {} discovered in the pending buffer, ignoring. Max consecutive: {}",
-                                           pendingEntry.epoch, prev.epoch));
+                             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()

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<?> accordReady = AccordService.instance().epochReadyFor(metadata);
-        Future<?> ready = FutureCombiner.allOf(Lists.newArrayList(bootstrapStream, accordReady));
+        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);

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)

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");

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);
     }
 

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;

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

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)

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();
+        }
+    }
+}

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
     {

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.junit.Assert;
-import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
+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)
-            instance.runOnInstance(() -> AccordService.unsafeSetNewAccordService(new BarrierRecordingService(AccordService.instance())));
-//        setupCluster(opt -> opt, 3);
     }
 
-    @After
+    private static Cluster createCluster() throws Throwable
-    public void tearDown()
     {
-        for (IInvokableInstance instance : SHARED_CLUSTER)
+        Cluster cluster = AccordTestBase.createCluster(3, opt -> opt.withConfig(conf -> conf.with(Feature.NETWORK, Feature.GOSSIP)
-            instance.runOnInstance(() -> AccordService.instance().node().commandStores().forEachCommandStore(cs -> cs.unsafeProgressLog().start()));
+                                                           .set("accord.recover_txn", "1s")
-        SHARED_CLUSTER.filters().reset();
+                                                           .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())));
+
+        return cluster;
     }
 
+    private static void withCluster(ModelChecker.ThrowingConsumer<Cluster> run) throws Throwable
+    {
+        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();
+        cluster.filters().allVerbs().to(3).drop();
-        awaitEndpointDown(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
+        awaitEndpointDown(cluster.get(1), 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" +
+                                                        "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.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();
+        cluster.filters().reset();
-        awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
+        awaitEndpointUp(cluster.get(1), cluster.get(3));
-        nodetool(SHARED_CLUSTER.get(1), "repair", KEYSPACE);
+        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);
+        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());
@@ -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();
+        cluster.filters().reset();
-        awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
+        awaitEndpointUp(cluster.get(1), cluster.get(3));
-        nodetool(SHARED_CLUSTER.get(1), "repair", "--accord-only", KEYSPACE);
+        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();
+        cluster.filters().allVerbs().to(3).drop();
-        awaitEndpointDown(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
+        awaitEndpointDown(cluster.get(1), cluster.get(3));
-        awaitEndpointDown(SHARED_CLUSTER.get(2), SHARED_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();
+        cluster.filters().reset();
-        awaitEndpointUp(SHARED_CLUSTER.get(1), SHARED_CLUSTER.get(3));
+        awaitEndpointUp(cluster.get(1), cluster.get(3));
-        nodetool(SHARED_CLUSTER.get(1), "repair", "--force", "--accord-only", KEYSPACE);
+        nodetool(cluster.get(1), "repair", "--force", "--accord-only", KEYSPACE);
 
-        SHARED_CLUSTER.get(1).runOnInstance(() -> {
+        cluster.get(1).runOnInstance(() -> {
             Assert.assertTrue(barrierRecordingService().executedBarriers);
         });
     }

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);
+        ClusterUtils.waitForCMSToQuiesce(SHARED_CLUSTER, maxEpoch(SHARED_CLUSTER, 1), true);
-        SHARED_CLUSTER.forEach(() -> Util.spinUntilTrue(() -> ClusterMetadata.current().epoch.getEpoch() ==
+        SHARED_CLUSTER.forEach(() -> Util.spinUntilTrue(() -> {
-                                                              ((AccordService) AccordService.instance()).configService().currentEpoch() &&
+            ClusterMetadata metadata = ClusterMetadata.current();
-                                                              AccordService.instance().topology().current().epoch() ==
+            if (!metadata.schema.hasAccordKeyspaces())
-                                                              ((AccordService) AccordService.instance()).configService().currentEpoch(),
+                return true;
-                                                        60));
+
+            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

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) -> {

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,

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()
     {

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))
+            if (keyspaceName.startsWith("BootStrapperTest"))
-                continue;
+                keyspaces.add(keyspaceName);
-            int replicationFactor = Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor().allReplicas;
-            for (int clusterSize : clusterSizes)
-                if (clusterSize >= replicationFactor)
-                    testSourceTargetComputation(keyspaceName, clusterSize, replicationFactor);
         }
+
+        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
+            s.addKeyspaceToFetch(keyspace);
-        // the RangeStreamer instance was constructed with a null Collection<Token>, relying on pending ranges being
+            Multimap<InetAddressAndPort, RangeStreamer.FetchReplica> toFetch = s.toFetch().get(keyspace);
-        // 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
+            // Pre-TCM this test would always run with RangeStreamer::useStrictSourcesForRanges returning false because
-        // is used, they will vary.
+            // the RangeStreamer instance was constructed with a null Collection<Token>, relying on pending ranges being
-        assert toFetch.values().size() > 0;
+            // 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());
 
-        assert toFetch.keys().stream().noneMatch(InetAddressAndPort.getByName("127.0.0.1")::equals);
+            // there isn't any point in testing the size of these collections for any specific size.  When a random partitioner
-        return s;
+            // is used, they will vary.
+            Assert.assertTrue(toFetch.values().size() > 0);
+
+            Assert.assertTrue(toFetch.keys().stream().noneMatch(myEndpoint::equals));
+        }
     }
 
     private RangeStreamer getRangeStreamer() throws UnknownHostException

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();
+            ClusterMetadata metadata = ClusterMetadata.current();
-            AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmNodeId));
+            AccordService.MetadataChangeListener.instance.resetForTesting(metadata);
-            Topology topology = new Topology(1, Shard.create(TokenRange.fullRange(tableId, getPartitioner()), ofSorted(new Node.Id(1)), ofSorted(new Node.Id(1))));
+            NodeId tcmNodeId = metadata.myNodeId();
-            as.unsafeStartupWithOverrides(new Journal.TopologyUpdate(storeRangesForEpochs, topology));
+            AccordService.unsafeSetNewAccordService(null);
-            for (CommandStore commandStore : as.node().commandStores().all())
+            return AccordService.startup(tcmNodeId);
-                ((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;
         }
 
         TxnId nextTxnId(Domain domain)
@@ -626,6 +608,7 @@ public class RouteIndexTest extends CQLTester
 
         private void restartAccord()
         {
+            accordService.journal().closeCurrentSegmentForTestingIfNonEmpty();
             accordService.shutdown();
             accordService = startAccord();
         }