Merge branch 'cassandra-6.0' into trunk

CHANGES.txt

@@ -3,6 +3,12 @@
  * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
  * Add a guardrail for misprepared statements (CASSANDRA-21139)
 Merged from 6.0:
+ * Don’t leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236)
+ * Make nodetool abortbootstrap more robust (CASSANDRA-21235)
+ * Don't clear prepared statement cache on nodetool cms initialize (CASSANDRA-21234)
+ * Improve performance when deserializing cluster metadata  (CASSANDRA-21224)
+ * Minor TokenMap performance improvement (CASSANDRA-21223)
+ * Handle lost response when committing PrepareMove (CASSANDRA-21222)
  * SEPExecutor.maybeExecuteImmediately does not always execute tasks immediately despite available worker capacity (CASSANDRA-21429)
  * Safely regain ranges and delete retired command stores (CASSANDRA-21212)
  * Reduce memory allocations in miscellaneous places along read path (CASSANDRA-21360)

src/java/org/apache/cassandra/gms/Gossiper.java

@@ -269,7 +269,13 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
                 taskLock.lock();
 
                 /* Update the local heartbeat counter. */
-                endpointStateMap.get(getBroadcastAddressAndPort()).updateHeartBeat();
+                EndpointState epstate = endpointStateMap.get(getBroadcastAddressAndPort());
+                if (epstate == null)
+                {
+                    logger.warn("Node {} is not in gossip, not running GossipTask", getBroadcastAddressAndPort());
+                    return;
+                }
+                epstate.updateHeartBeat();
                 if (logger.isTraceEnabled())
                     logger.trace("My heartbeat is now {}", endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().getHeartBeatVersion());
                 final List<GossipDigest> gDigests = new ArrayList<>();

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

@@ -392,6 +392,11 @@ public class CassandraDaemon
         // Prepared statements
         QueryProcessor.instance.preloadPreparedStatements();
 
+        // Apply overrides before re-enabling auto-compaction
+        setCompactionStrategyOverrides(Schema.instance.getKeyspaces());
+        // re-enable auto-compaction after replay, so correct disk boundaries are used
+        enableAutoCompaction(Schema.instance.getKeyspaces());
+
         // start server internals
         StorageService.instance.registerDaemon(this);
         try
@@ -424,11 +429,6 @@ public class CassandraDaemon
         ScheduledExecutors.optionalTasks.schedule(viewRebuild, StorageService.RING_DELAY_MILLIS, TimeUnit.MILLISECONDS);
         StorageService.instance.doAuthSetup();
 
-        // Apply overrides before re-enabling auto-compaction
-        setCompactionStrategyOverrides(Schema.instance.getKeyspaces());
-        // re-enable auto-compaction after replay, so correct disk boundaries are used
-        enableAutoCompaction(Schema.instance.getKeyspaces());
-
         AuditLogManager.instance.initialize();
 
         StorageService.instance.doAutoRepairSetup();

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

@@ -1663,15 +1663,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
 
     public void abortBootstrap(String nodeStr, String endpointStr)
     {
-        logger.debug("Aborting bootstrap for {}/{}", nodeStr, endpointStr);
+        logger.info("Aborting bootstrap for {}", StringUtils.isEmpty(nodeStr) ? endpointStr : nodeStr);
         ClusterMetadata metadata = ClusterMetadata.current();
-        NodeId nodeId;
+        NodeId nodeId = parseNodeIdOrEndpoint(metadata, nodeStr, endpointStr);
-        if (!StringUtils.isEmpty(nodeStr))
-            nodeId = NodeId.fromString(nodeStr);
-        else
-            nodeId = metadata.directory.peerId(InetAddressAndPort.getByNameUnchecked(endpointStr));
-
         InetAddressAndPort endpoint = metadata.directory.endpoint(nodeId);
+        if (endpoint == null)
+            throw new IllegalArgumentException("Can't abort bootstrap for " + nodeId + " - it does not exist in cluster metadata");
         if (Gossiper.instance.isKnownEndpoint(endpoint) && FailureDetector.instance.isAlive(endpoint))
             throw new RuntimeException("Can't abort bootstrap for " + nodeId + " - it is alive");
         NodeState nodeState = metadata.directory.peerState(nodeId);
@@ -1694,6 +1691,47 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         }
     }
 
+    private static NodeId parseNodeIdOrEndpoint(ClusterMetadata metadata, String nodeStr, String endpointStr)
+    {
+        NodeId nodeId;
+        if (!StringUtils.isEmpty(nodeStr))
+        {
+            try
+            {
+                nodeId = NodeId.fromString(nodeStr);
+            }
+            catch (IllegalArgumentException | UnsupportedOperationException e)
+            {
+                String msg = "Unable to parse node id string " + nodeStr;
+                logger.warn("{}", msg, e);
+                throw new IllegalArgumentException(msg, e);
+            }
+        }
+        else
+        {
+            InetAddressAndPort endpoint;
+            try
+            {
+                endpoint = InetAddressAndPort.getByName(endpointStr);
+            }
+            catch (UnknownHostException e)
+            {
+                String msg = "Unable to look up endpoint " + endpointStr;
+                logger.warn("{}", msg, e);
+                throw new IllegalArgumentException(msg, e);
+            }
+
+            nodeId = metadata.directory.peerId(endpoint);
+            if (nodeId == null)
+            {
+                String msg = "Unknown endpoint: " + endpoint;
+                logger.warn(msg);
+                throw new IllegalArgumentException(msg);
+            }
+        }
+        return nodeId;
+    }
+
     @Override
     public void migrateConsensusProtocol(@Nonnull List<String> keyspaceNames,
                                          @Nullable List<String> maybeTableNames,
@@ -2010,7 +2048,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
 
     public String getLocalHostId()
     {
-        return getLocalHostUUID().toString();
+        UUID localHostId = getLocalHostUUID();
+        return localHostId != null ? localHostId.toString() : "UNKNOWN";
     }
 
     public UUID getLocalHostUUID()

src/java/org/apache/cassandra/tcm/ClusterMetadata.java

@@ -1062,11 +1062,6 @@ public class ClusterMetadata
         return null;
     }
 
-    public boolean metadataSerializationUpgradeInProgress()
-    {
-        return !directory.clusterMaxVersion.serializationVersion().equals(directory.commonSerializationVersion);
-    }
-
     public static class Serializer implements MetadataSerializer<ClusterMetadata>
     {
         @Override

src/java/org/apache/cassandra/tcm/Startup.java

@@ -449,6 +449,8 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
                 if (isReplacing)
                     ReconfigureCMS.maybeReconfigureCMS(metadata, DatabaseDescriptor.getReplaceAddress());
 
+                // if this throws startup is aborted and operator needs to restart, in that case the IPS is resumed if
+                // it was successfully committed
                 ClusterMetadataService.instance().commit(initialTransformation.get());
                 // 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

src/java/org/apache/cassandra/tcm/listeners/InitializationListener.java

@@ -1,36 +0,0 @@
-/*
- * 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.tcm.listeners;
-
-import org.apache.cassandra.cql3.QueryProcessor;
-import org.apache.cassandra.tcm.Transformation;
-import org.apache.cassandra.tcm.log.Entry;
-
-public class InitializationListener implements LogListener
-{
-    @Override
-    public void notify(Entry entry, Transformation.Result result)
-    {
-        if (entry.transform.kind() == Transformation.Kind.INITIALIZE_CMS)
-        {
-            QueryProcessor.clearPreparedStatementsCache();
-            QueryProcessor.clearInternalStatementsCache();
-        }
-    }
-}

src/java/org/apache/cassandra/tcm/log/LocalLog.java

@@ -57,7 +57,6 @@ import org.apache.cassandra.tcm.Startup;
 import org.apache.cassandra.tcm.Transformation;
 import org.apache.cassandra.tcm.listeners.ChangeListener;
 import org.apache.cassandra.tcm.listeners.ClientNotificationListener;
-import org.apache.cassandra.tcm.listeners.InitializationListener;
 import org.apache.cassandra.tcm.listeners.LegacyStateListener;
 import org.apache.cassandra.tcm.listeners.LogListener;
 import org.apache.cassandra.tcm.listeners.MetadataSnapshotListener;
@@ -905,7 +904,6 @@ public abstract class LocalLog implements Closeable
         changeListeners.clear();
 
         addListener(snapshotListener());
-        addListener(new InitializationListener());
         addListener(new SchemaListener(spec.loadSSTables));
         addListener(new LegacyStateListener());
         addListener(new PlacementsChangeListener());

src/java/org/apache/cassandra/tcm/membership/Directory.java

@@ -50,7 +50,6 @@ import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tcm.MetadataValue;
 import org.apache.cassandra.tcm.serialization.MetadataSerializer;
 import org.apache.cassandra.tcm.serialization.Version;
-import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.UUIDSerializer;
 import org.apache.cassandra.utils.btree.BTreeBiMap;
 import org.apache.cassandra.utils.btree.BTreeMap;
@@ -58,6 +57,7 @@ import org.apache.cassandra.utils.btree.BTreeMultimap;
 
 import static org.apache.cassandra.db.TypeSizes.sizeof;
 import static org.apache.cassandra.tcm.membership.NodeVersion.CURRENT;
+import static org.apache.cassandra.tcm.membership.NodeVersion.CURRENT_METADATA_VERSION;
 
 public class Directory implements MetadataValue<Directory>
 {
@@ -106,6 +106,41 @@ public class Directory implements MetadataValue<Directory>
                       BTreeMap<NodeId, NodeAddresses> addresses,
                       BTreeMultimap<String, InetAddressAndPort> endpointsByDC,
                       BTreeMap<String, Multimap<String, InetAddressAndPort>> racksByDC)
+    {
+        this(nextId, lastModified, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC, racksByDC, clusterVersions(states, versions));
+    }
+
+    private Directory(int nextId,
+                      Epoch lastModified,
+                      BTreeBiMap<NodeId, InetAddressAndPort> peers,
+                      BTreeSet<RemovedNode> removedNodes,
+                      BTreeMap<NodeId, Location> locations,
+                      BTreeMap<NodeId, NodeState> states,
+                      BTreeMap<NodeId, NodeVersion> versions,
+                      BTreeBiMap<NodeId, UUID> hostIds,
+                      BTreeMap<NodeId, NodeAddresses> addresses,
+                      BTreeMultimap<String, InetAddressAndPort> endpointsByDC,
+                      BTreeMap<String, Multimap<String, InetAddressAndPort>> racksByDC,
+                      ClusterVersions clusterVersions)
+    {
+        this(nextId, lastModified, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC, racksByDC,
+             clusterVersions.clusterMinVersion, clusterVersions.clusterMaxVersion, clusterVersions.commonSerializationVersion);
+    }
+
+    private Directory(int nextId,
+                      Epoch lastModified,
+                      BTreeBiMap<NodeId, InetAddressAndPort> peers,
+                      BTreeSet<RemovedNode> removedNodes,
+                      BTreeMap<NodeId, Location> locations,
+                      BTreeMap<NodeId, NodeState> states,
+                      BTreeMap<NodeId, NodeVersion> versions,
+                      BTreeBiMap<NodeId, UUID> hostIds,
+                      BTreeMap<NodeId, NodeAddresses> addresses,
+                      BTreeMultimap<String, InetAddressAndPort> endpointsByDC,
+                      BTreeMap<String, Multimap<String, InetAddressAndPort>> racksByDC,
+                      NodeVersion clusterMinVersion,
+                      NodeVersion clusterMaxVersion,
+                      Version commonSerializationVersion)
     {
         this.nextId = nextId;
         this.lastModified = lastModified;
@@ -118,10 +153,9 @@ public class Directory implements MetadataValue<Directory>
         this.addresses = addresses;
         this.endpointsByDC = endpointsByDC;
         this.racksByDC = racksByDC;
-        Pair<NodeVersion, NodeVersion> minMaxVer = minMaxVersions(states, versions);
+        this.clusterMinVersion = clusterMinVersion;
-        clusterMinVersion = minMaxVer.left;
+        this.clusterMaxVersion = clusterMaxVersion;
-        clusterMaxVersion = minMaxVer.right;
+        this.commonSerializationVersion = commonSerializationVersion;
-        commonSerializationVersion = minCommonSerializationVersion(states, versions);
     }
 
     @Override
@@ -161,7 +195,7 @@ public class Directory implements MetadataValue<Directory>
     @Override
     public Directory withLastModified(Epoch epoch)
     {
-        return new Directory(nextId, epoch, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC, racksByDC);
+        return new Directory(nextId, epoch, peers, removedNodes, locations, states, versions, hostIds, addresses, endpointsByDC, racksByDC, clusterMinVersion, clusterMaxVersion, commonSerializationVersion);
     }
 
     public Directory withNonUpgradedNode(NodeAddresses addresses,
@@ -250,9 +284,18 @@ public class Directory implements MetadataValue<Directory>
         BTreeMap<String, Multimap<String, InetAddressAndPort>> updatedEndpointsByRack = racksByDC.withForce(location(id).datacenter, rackEP);
 
         return new Directory(nextId, lastModified,
-                             peers.withForce(id,nodeAddresses.broadcastAddress), removedNodes, locations, states, versions, hostIds, addresses.withForce(id, nodeAddresses),
+                             peers.withForce(id, nodeAddresses.broadcastAddress),
+                             removedNodes,
+                             locations,
+                             states,
+                             versions,
+                             hostIds,
+                             addresses.withForce(id, nodeAddresses),
                              updatedEndpointsByDC,
-                             updatedEndpointsByRack);
+                             updatedEndpointsByRack,
+                             clusterMinVersion,
+                             clusterMaxVersion,
+                             commonSerializationVersion);
     }
 
     public Directory withRackAndDC(NodeId id)
@@ -266,7 +309,10 @@ public class Directory implements MetadataValue<Directory>
 
         return new Directory(nextId, lastModified, peers, removedNodes, locations, states, versions, hostIds, addresses,
                              endpointsByDC.with(location.datacenter, endpoint),
-                             racksByDC.withForce(location.datacenter, rackEP));
+                             racksByDC.withForce(location.datacenter, rackEP),
+                             clusterMinVersion,
+                             clusterMaxVersion,
+                             commonSerializationVersion);
     }
 
     public Directory withoutRackAndDC(NodeId id)
@@ -286,7 +332,10 @@ public class Directory implements MetadataValue<Directory>
             newRacksByDC = racksByDC.withForce(location.datacenter, rackEP);
         return new Directory(nextId, lastModified, peers, removedNodes, locations, states, versions, hostIds, addresses,
                              endpointsByDC.without(location.datacenter, endpoint),
-                             newRacksByDC);
+                             newRacksByDC,
+                             clusterMinVersion,
+                             clusterMaxVersion,
+                             commonSerializationVersion);
     }
 
     public Directory withUpdatedRackAndDc(NodeId id, Location location)
@@ -306,23 +355,7 @@ public class Directory implements MetadataValue<Directory>
             return this;
 
         return new Directory(nextId, lastModified, peers, removedNodes, locations.withForce(id, location), states, versions, hostIds,
-                             addresses, endpointsByDC, racksByDC);
+                             addresses, endpointsByDC, racksByDC, clusterMinVersion, clusterMaxVersion, commonSerializationVersion);
-    }
-
-    public Directory removed(Epoch removedIn, NodeId id, InetAddressAndPort addr)
-    {
-        Invariants.require(!peers.containsKey(id));
-        return new Directory(nextId,
-                             lastModified,
-                             peers,
-                             removedNodes.with(new RemovedNode(removedIn, id, addr)),
-                             locations,
-                             states,
-                             versions,
-                             hostIds,
-                             addresses,
-                             endpointsByDC,
-                             racksByDC);
     }
 
     public Directory without(Epoch removedIn, NodeId id)
@@ -641,14 +674,23 @@ public class Directory implements MetadataValue<Directory>
             if (version.isAtLeast(Version.V1))
                 nextId = in.readInt();
             int count = in.readInt();
-            Directory newDir = new Directory();
 
+            BTreeBiMap<NodeId, InetAddressAndPort> peers = BTreeBiMap.empty();
+            BTreeMap<NodeId, Location> locations = BTreeMap.empty();
+            BTreeMap<NodeId, NodeState> states = BTreeMap.empty();
+            BTreeMap<NodeId, NodeVersion> versions = BTreeMap.empty();
+            BTreeBiMap<NodeId, UUID> hostIds = BTreeBiMap.empty();
+            BTreeMap<NodeId, NodeAddresses> addresses = BTreeMap.empty();
             for (int i = 0; i < count; i++)
             {
                 Node n = Node.serializer.deserialize(in, version);
-                // todo: bulk operations
+                NodeId id = n.id;
-                newDir = newDir.with(n.addresses, n.id, n.hostId, n.location, n.version)
+                peers = peers.withForce(id, n.addresses.broadcastAddress);
-                               .withNodeState(n.id, n.state);
+                locations = locations.withForce(id, n.location);
+                states = states.withForce(id, n.state);
+                versions = versions.withForce(id, n.version);
+                hostIds = hostIds.withForce(id, n.hostId);
+                addresses = addresses.withForce(id, n.addresses);
             }
 
             int dcCount = in.readInt();
@@ -677,7 +719,7 @@ public class Directory implements MetadataValue<Directory>
             if (version.isBefore(Version.V1))
             {
                 NodeId maxId = null;
-                for (NodeId id : newDir.peers.keySet())
+                for (NodeId id : peers.keySet())
                 {
                     if (maxId == null || id.compareTo(maxId) > 0)
                         maxId = id;
@@ -688,7 +730,7 @@ public class Directory implements MetadataValue<Directory>
                 else
                     nextId = maxId.id() + 1;
             }
-
+            BTreeSet<RemovedNode> removed = BTreeSet.empty(RemovedNode::compareTo);
             if (version.isAtLeast(Version.V7))
             {
                 int removedNodes = in.readInt();
@@ -697,19 +739,20 @@ public class Directory implements MetadataValue<Directory>
                     long epoch = in.readLong();
                     NodeId nodeId = NodeId.serializer.deserialize(in, version);
                     InetAddressAndPort addr = InetAddressAndPort.MetadataSerializer.serializer.deserialize(in, version);
-                    newDir.removed(Epoch.create(epoch), nodeId, addr);
+                    Invariants.require(!peers.containsKey(nodeId));
+                    removed = removed.with(new RemovedNode(Epoch.create(epoch), nodeId, addr));
                 }
             }
 
             return new Directory(nextId,
                                  lastModified,
-                                 newDir.peers,
+                                 peers,
-                                 newDir.removedNodes,
+                                 removed,
-                                 newDir.locations,
+                                 locations,
-                                 newDir.states,
+                                 states,
-                                 newDir.versions,
+                                 versions,
-                                 newDir.hostIds,
+                                 hostIds,
-                                 newDir.addresses,
+                                 addresses,
                                  dcEndpoints,
                                  racksByDC);
         }
@@ -769,10 +812,11 @@ public class Directory implements MetadataValue<Directory>
                equivalentTo(directory);
     }
 
-    private static Pair<NodeVersion, NodeVersion> minMaxVersions(BTreeMap<NodeId, NodeState> states, BTreeMap<NodeId, NodeVersion> versions)
+    private static ClusterVersions clusterVersions(BTreeMap<NodeId, NodeState> states, BTreeMap<NodeId, NodeVersion> versions)
     {
         NodeVersion minVersion = null;
         NodeVersion maxVersion = null;
+        int commonVersion = Integer.MAX_VALUE;
         for (Map.Entry<NodeId, NodeState> entry : states.entrySet())
         {
             if (entry.getValue() != NodeState.LEFT)
@@ -782,26 +826,15 @@ public class Directory implements MetadataValue<Directory>
                     minVersion = ver;
                 if (maxVersion == null || ver.compareTo(maxVersion) > 0)
                     maxVersion = ver;
-            }
-        }
-        if (minVersion == null)
-            return Pair.create(CURRENT, CURRENT);
-        return Pair.create(minVersion, maxVersion);
-    }
-
-    public static Version minCommonSerializationVersion(BTreeMap<NodeId, NodeState> states, BTreeMap<NodeId, NodeVersion> versions)
-    {
-        int commonVersion = Integer.MAX_VALUE;
-        for (Map.Entry<NodeId, NodeState> entry : states.entrySet())
-        {
-            if (entry.getValue() != NodeState.LEFT)
-            {
-                NodeVersion ver = versions.get(entry.getKey());
                 if (ver.serializationVersion > Version.OLD.asInt() && ver.serializationVersion < commonVersion)
                     commonVersion = ver.serializationVersion;
             }
         }
-        return commonVersion == Integer.MAX_VALUE ? NodeVersion.CURRENT_METADATA_VERSION : Version.fromInt(commonVersion);
+        if (minVersion == null)
+            return new ClusterVersions(CURRENT, CURRENT, CURRENT_METADATA_VERSION);
+
+        return new ClusterVersions(minVersion, maxVersion,
+                                   commonVersion == Integer.MAX_VALUE ? NodeVersion.CURRENT_METADATA_VERSION : Version.fromInt(commonVersion));
     }
 
     @Override
@@ -825,7 +858,8 @@ public class Directory implements MetadataValue<Directory>
                Objects.equals(endpointsByDC, directory.endpointsByDC) &&
                Objects.equals(racksByDC, directory.racksByDC) &&
                Objects.equals(versions, directory.versions) &&
-               Objects.equals(addresses, directory.addresses);
+               Objects.equals(addresses, directory.addresses) &&
+               Objects.equals(removedNodes, directory.removedNodes);
     }
     
     private static final Logger logger = LoggerFactory.getLogger(Directory.class);
@@ -891,7 +925,6 @@ public class Directory implements MetadataValue<Directory>
 
     }
 
-
     public static class RemovedNode implements Comparable<RemovedNode>
     {
         public final Epoch removedIn;
@@ -923,4 +956,18 @@ public class Directory implements MetadataValue<Directory>
             return id.compareTo(o.id);
         }
     }
+
+    private static class ClusterVersions
+    {
+        private final NodeVersion clusterMinVersion;
+        private final NodeVersion clusterMaxVersion;
+        private final Version commonSerializationVersion;
+        public ClusterVersions(NodeVersion clusterMinVersion, NodeVersion clusterMaxVersion, Version commonSerializationVersion)
+        {
+
+            this.clusterMinVersion = clusterMinVersion;
+            this.clusterMaxVersion = clusterMaxVersion;
+            this.commonSerializationVersion = commonSerializationVersion;
+        }
+    }
 }

src/java/org/apache/cassandra/tcm/ownership/TokenMap.java

@@ -55,7 +55,7 @@ public class TokenMap implements MetadataValue<TokenMap>
     private static final Logger logger = LoggerFactory.getLogger(TokenMap.class);
 
     private final SortedBiMultiValMap<Token, NodeId> map;
-    private final List<Token> tokens;
+    private final ImmutableList<Token> tokens;
     private final List<Range<Token>> ranges;
     // TODO: move partitioner to the users (SimpleStrategy and Uniform Range Placement?)
     private final IPartitioner partitioner;
@@ -71,7 +71,7 @@ public class TokenMap implements MetadataValue<TokenMap>
         this.lastModified = lastModified;
         this.partitioner = partitioner;
         this.map = map;
-        this.tokens = tokens();
+        this.tokens = ImmutableList.copyOf(map.keySet());
         this.ranges = toRanges(tokens, partitioner);
     }
 
@@ -101,18 +101,6 @@ public class TokenMap implements MetadataValue<TokenMap>
         return new TokenMap(lastModified, partitioner, finalisedCopy);
     }
 
-    public TokenMap unassignTokens(NodeId id, Collection<Token> tokens)
-    {
-        SortedBiMultiValMap<Token, NodeId> finalisedCopy = SortedBiMultiValMap.create(map);
-        for (Token token : tokens)
-        {
-            NodeId nodeId = finalisedCopy.remove(token);
-            assert nodeId.equals(id);
-        }
-
-        return new TokenMap(lastModified, partitioner, finalisedCopy);
-    }
-
     public SortedBiMultiValMap<Token, NodeId> asMap()
     {
         return SortedBiMultiValMap.create(map);
@@ -130,7 +118,7 @@ public class TokenMap implements MetadataValue<TokenMap>
 
     public ImmutableList<Token> tokens()
     {
-        return ImmutableList.copyOf(map.keySet());
+        return tokens;
     }
 
     public ImmutableList<Token> tokens(NodeId nodeId)

src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java

@@ -85,7 +85,9 @@ public interface SingleNodeSequences
             ClusterMetadataService.instance().commit(new PrepareLeave(self,
                                                                       force,
                                                                       ClusterMetadataService.instance().placementProvider(),
-                                                                      LeaveStreams.Kind.UNBOOTSTRAP));
+                                                                      LeaveStreams.Kind.UNBOOTSTRAP),
+                                                     m -> m,
+                                                     failureHandler("PrepareLeave", StorageService.instance::markDecommissionFailed));
         }
         else if (InProgressSequences.isLeave(inProgress))
         {
@@ -182,13 +184,24 @@ public interface SingleNodeSequences
         ClusterMetadataService.instance().commit(new PrepareMove(self,
                                                                  Collections.singleton(newToken),
                                                                  ClusterMetadataService.instance().placementProvider(),
-                                                                 true));
+                                                                 true),
+                                                 m -> m,
+                                                 failureHandler("PrepareMove", StorageService.instance::markMoveFailed));
         InProgressSequences.finishInProgressSequences(self);
 
         if (logger.isDebugEnabled())
             logger.debug("Successfully moved to new token {}", StorageService.instance.getLocalTokens().iterator().next());
     }
 
+    private static ClusterMetadataService.CommitFailureHandler<ClusterMetadata> failureHandler(String type, Runnable markFailed)
+    {
+        return (code, msg) -> {
+            logger.warn("Got failure committing {} transformation: {} {}", type, code, msg);
+            markFailed.run();
+            throw new IllegalStateException(String.format("Can not commit transformation: \"%s\"(%s).", code, msg));
+        };
+    }
+
     static void resumeMove()
     {
         if (ClusterMetadataService.instance().isMigrating() || ClusterMetadataService.state() == ClusterMetadataService.State.GOSSIP)
@@ -201,6 +214,11 @@ public interface SingleNodeSequences
         {
             String msg = "No move operation in progress, can't resume";
             logger.info(msg);
+            if (StorageService.instance.operationMode() == MOVE_FAILED)
+            {
+                // there is no ongoing move to resume, but operation mode thinks there is
+                StorageService.instance.clearTransientMode();
+            }
             throw new IllegalStateException(msg);
         }
         if (StorageService.instance.operationMode() != MOVE_FAILED)
@@ -221,7 +239,7 @@ public interface SingleNodeSequences
     /**
      *
      * @param nodeId node id to abort the MSO for, null for local node
-     * @param kind the expected kind of the multi step operation to abolt
+     * @param kind the expected kind of the multi step operation to abort
      * @param ssMode the legacy mode we want storage service to be in, null for any
      */
     private static void abortHelper(@Nullable String nodeId, MultiStepOperation.Kind kind, @Nullable StorageService.Mode ssMode)
@@ -234,9 +252,19 @@ public interface SingleNodeSequences
         MultiStepOperation<?> sequence = metadata.inProgressSequences.get(toAbort);
         if (sequence == null || sequence.kind() != kind)
         {
-            String msg = String.format("No %s operation in progress for %s, can't abort (%s)", kind, toAbort, sequence);
+            if (toAbort.equals(metadata.myNodeId()) && ssMode != null && StorageService.instance.operationMode() == ssMode)
-            logger.info(msg);
+            {
-            throw new IllegalStateException(msg);
+                // there is no ongoing sequence with the given kind, but storage service operation mode is set, clear it
+                logger.debug("There is no ongoing {} sequence for this node, but operation mode is {} - clearing transient mode", kind, ssMode);
+                StorageService.instance.clearTransientMode();
+                return;
+            }
+            else
+            {
+                String msg = String.format("No %s operation in progress for %s, can't abort (%s)", kind, toAbort, sequence);
+                logger.info(msg);
+                throw new IllegalStateException(msg);
+            }
         }
         if (toAbort.equals(metadata.myNodeId()))
         {

src/java/org/apache/cassandra/tcm/sequences/UnbootstrapAndLeave.java

@@ -92,12 +92,12 @@ public class UnbootstrapAndLeave extends MultiStepOperation<Epoch>
      */
     @VisibleForTesting
     UnbootstrapAndLeave(Epoch latestModification,
-                               LockedRanges.Key lockKey,
+                        LockedRanges.Key lockKey,
-                               Transformation.Kind next,
+                        Transformation.Kind next,
-                               PrepareLeave.StartLeave startLeave,
+                        PrepareLeave.StartLeave startLeave,
-                               PrepareLeave.MidLeave midLeave,
+                        PrepareLeave.MidLeave midLeave,
-                               PrepareLeave.FinishLeave finishLeave,
+                        PrepareLeave.FinishLeave finishLeave,
-                               LeaveStreams streams)
+                        LeaveStreams streams)
     {
         super(nextToIndex(next), latestModification);
         this.lockKey = lockKey;
@@ -198,7 +198,8 @@ public class UnbootstrapAndLeave extends MultiStepOperation<Epoch>
                 }
                 catch (ExecutionException e)
                 {
-                    StorageService.instance.markDecommissionFailed();
+                    if (startLeave.nodeId().equals(ClusterMetadata.current().myNodeId()))
+                        StorageService.instance.markDecommissionFailed();
                     JVMStabilityInspector.inspectThrowable(e);
                     logger.error("Error while decommissioning node: {}", e.getCause().getMessage());
                     throw new RuntimeException("Error while decommissioning node: " + e.getCause().getMessage());

src/java/org/apache/cassandra/tcm/transformations/Register.java

@@ -176,6 +176,9 @@ public class Register implements Transformation
         else
         {
             NodeId nodeId = directory.peerId(FBUtilities.getBroadcastAddressAndPort());
+            if (nodeId == null)
+                throw new IllegalStateException("Node has host id "+localHostId+" in system.local, but is not present in cluster metadata - not allowing this node to register. " +
+                                                "If a bootstrap of this node failed and was aborted with `nodetool abortbootstrap` it should also have its data removed before trying to rebootstrap.");
             NodeVersion dirVersion = directory.version(nodeId);
 
             // If this is a node in the process of upgrading, update the host id in the system.local table

test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapCompactionTest.java

@@ -0,0 +1,93 @@
+/*
+ * 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.ring;
+
+import java.util.concurrent.Callable;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.Constants;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.tcm.sequences.BootstrapAndJoin;
+import org.apache.cassandra.tcm.sequences.SequenceState;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.junit.Assert.assertTrue;
+
+public class BootstrapCompactionTest extends TestBaseImpl
+{
+    @Test
+    public void testCompactionEnabledDuringBootstrap() throws Exception
+    {
+        int originalNodeCount = 2;
+        int expandedNodeCount = originalNodeCount + 1;
+
+        try (Cluster cluster = init(builder().withNodes(originalNodeCount)
+                                             .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(expandedNodeCount))
+                                             .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0"))
+                                             .withInstanceInitializer(BB::install)
+                                             .withConfig(config -> config.with(NETWORK, GOSSIP))
+                                             .start()))
+        {
+            cluster.schemaChange(withKeyspace("create table %s.tbl (id int primary key)"));
+            IInstanceConfig config = cluster.newInstanceConfig()
+                                            .set(Constants.KEY_DTEST_FULL_STARTUP, true)
+                                            .set("auto_bootstrap", true);
+
+            IInvokableInstance newInstance = cluster.bootstrap(config);
+            // BB below asserts that autocompaction is enabled at each step in the join sequence
+            newInstance.startup(cluster);
+        }
+    }
+
+    public static class BB
+    {
+        public static void install(ClassLoader cl, int i)
+        {
+            if (i == 3)
+            {
+                new ByteBuddy().rebase(BootstrapAndJoin.class)
+                               .method(named("executeNext"))
+                               .intercept(MethodDelegation.to(BB.class))
+                               .make()
+                               .load(cl, ClassLoadingStrategy.Default.INJECTION);
+            }
+        }
+
+        public static SequenceState executeNext(@SuperCall Callable<SequenceState> zuper) throws Exception
+        {
+            boolean isEnabled = Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").getCompactionStrategyManager().isEnabled();
+            assertTrue("Autocompaction should be enabled during the bootstrap", isEnabled);
+            return zuper.call();
+        }
+    }
+}

test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java

@@ -18,6 +18,8 @@
 
 package org.apache.cassandra.distributed.test.ring;
 
+import java.io.Closeable;
+import java.io.IOException;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.Callable;
@@ -48,6 +50,7 @@ import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.distributed.api.ICluster;
 import org.apache.cassandra.distributed.api.IInstanceConfig;
 import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.NodeToolResult;
 import org.apache.cassandra.distributed.api.TokenSupplier;
 import org.apache.cassandra.distributed.shared.JMXUtil;
 import org.apache.cassandra.distributed.shared.NetworkTopology;
@@ -55,7 +58,6 @@ import org.apache.cassandra.distributed.test.TestBaseImpl;
 import org.apache.cassandra.metrics.DefaultNameFactory;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.service.StorageServiceMBean;
-import org.apache.cassandra.utils.Closeable;
 import org.apache.cassandra.utils.concurrent.CountDownLatch;
 
 import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -228,6 +230,26 @@ public class BootstrapTest extends TestBaseImpl
         }
     }
 
+    @Test
+    public void testAbortBootstrapBadIp() throws IOException
+    {
+        try (Cluster cluster = builder().withNodes(1).start())
+        {
+            NodeToolResult res = cluster.get(1).nodetoolResult("abortbootstrap", "--ip", "127.0.0.55");
+            res.asserts().failure();
+            assertTrue(res.getStdout().contains("Unknown endpoint"));
+            res = cluster.get(1).nodetoolResult("abortbootstrap", "--ip", "127.0.0.999");
+            res.asserts().failure();
+            assertTrue(res.getStdout().contains("Unable to look up endpoint"));
+            res = cluster.get(1).nodetoolResult("abortbootstrap", "--node", "999");
+            res.asserts().failure();
+            assertTrue(res.getStdout().contains("does not exist in cluster metadata"));
+            res = cluster.get(1).nodetoolResult("abortbootstrap", "--node", "hello");
+            res.asserts().failure();
+            assertTrue(res.getStdout().contains("Unable to parse node id"));
+        }
+    }
+
     public static void populate(ICluster cluster, int from, int to)
     {
         populate(cluster, from, to, 1, 3, ConsistencyLevel.QUORUM);

test/distributed/org/apache/cassandra/distributed/test/tcm/ClusterMetadataAbortedBootstrapRejoinTest.java

@@ -0,0 +1,109 @@
+/*
+ * 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.tcm;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.distributed.shared.NetworkTopology;
+import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.service.StorageService;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+
+public class ClusterMetadataAbortedBootstrapRejoinTest extends TestBaseImpl
+{
+    @Test
+    public void testFailedBootstrapNotAllowedToJoin() throws IOException, TimeoutException, ExecutionException, InterruptedException
+    {
+        TokenSupplier even = TokenSupplier.evenlyDistributedTokens(3);
+        try (Cluster cluster = init(Cluster.build(2)
+                                           .withInstanceInitializer(BBHelper::install)
+                                           .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
+                                           .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(3, "dc0", "rack0"))
+                                           .withTokenSupplier(even::token)
+                                           .start()))
+        {
+            IInstanceConfig config = cluster.newInstanceConfig()
+                                            .set("auto_bootstrap", true);
+            IInvokableInstance toBootstrap = cluster.bootstrap(config);
+            toBootstrap.startup(cluster);
+            toBootstrap.logs().watchFor(Duration.ofSeconds(60), BBHelper.FAILMESSAGE);
+            toBootstrap.shutdown().get();
+            cluster.get(1).runOnInstance(() -> {
+                int i = 0;
+                while (FailureDetector.instance.isAlive(InetAddressAndPort.getByNameUnchecked("127.0.0.3")) && i++ < 30)
+                    Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
+            });
+            cluster.get(1).nodetoolResult("abortbootstrap", "--ip", "127.0.0.3").asserts().success();
+            Assertions.assertThatThrownBy(toBootstrap::startup)
+                      .isInstanceOf(IllegalStateException.class)
+                      .hasMessageContaining("but is not present in cluster metadata");
+        }
+    }
+
+    public static class BBHelper
+    {
+        public static String FAILMESSAGE = "ARTIFICIALLY FAILING BOOTSTRAP";
+        public static AtomicBoolean enabled = new AtomicBoolean(true);
+        public static void install(ClassLoader cl, int i)
+        {
+            if (i == 3)
+            {
+                new ByteBuddy().rebase(StorageService.class)
+                               .method(named("repairPaxosForTopologyChange").and(takesArguments(1)))
+                               .intercept(MethodDelegation.to(BBHelper.class))
+                               .make()
+                               .load(cl, ClassLoadingStrategy.Default.INJECTION);
+            }
+        }
+
+        public static void repairPaxosForTopologyChange(String reason)
+        {
+            if (enabled.get())
+            {
+                enabled.set(false);
+                throw new RuntimeException(FAILMESSAGE);
+            }
+        }
+
+    }
+}

test/distributed/org/apache/cassandra/distributed/test/tcm/LostCommitReqResTest.java

@@ -0,0 +1,135 @@
+/*
+ * 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.tcm;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+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.test.TestBaseImpl;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.service.StorageService;
+
+import static org.junit.Assert.assertEquals;
+
+public class LostCommitReqResTest extends TestBaseImpl
+{
+    @Test
+    public void lostMoveCommitResponseTest() throws IOException
+    {
+        try (Cluster cluster = init(builder().withNodes(2)
+                                             .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP).set("cms_await_timeout", "1s").set("cms_default_max_retries", "5"))
+                                             .start()))
+        {
+            // no commit responses
+            cluster.filters().verbs(Verb.TCM_COMMIT_RSP.id).from(1).to(2).drop();
+            // lost response when committing PrepareMove, fails the nodetool command and halts progress
+            cluster.get(2).nodetoolResult("move", "1234").asserts().failure();
+            assertMoveFailed(cluster.get(2)); // we should be in MOVE_FAILED state to allow abortmove
+            // still no responses, committing CancelInProgressSequence response is lost, but is actually committed
+            cluster.get(2).nodetoolResult("abortmove").asserts().failure();
+            assertNormal(cluster.get(2)); // and we should be back to normal
+            cluster.get(2).nodetoolResult("move", "1234").asserts().failure();
+            assertMoveFailed(cluster.get(2));
+            // finishing the MSO does not depend on any commit responses, just that ClusterMetadata.current() is up to date, so this is successful;
+            cluster.get(2).nodetoolResult("move", "--resume").asserts().success();
+            assertNormal(cluster.get(2));
+        }
+    }
+
+    @Test
+    public void lostMoveCommitRequestTest() throws IOException
+    {
+        try (Cluster cluster = init(builder().withNodes(2)
+                                             .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP).set("cms_await_timeout", "1s").set("cms_default_max_retries", "5"))
+                                             .start()))
+        {
+            // no commit requests
+            cluster.filters().verbs(Verb.TCM_COMMIT_REQ.id).from(2).to(1).drop();
+            cluster.get(2).nodetoolResult("move", "1234").asserts().failure();
+            // state should be "move failed" since we don't know if the request or response went missing:
+            assertMoveFailed(cluster.get(2));
+            cluster.filters().reset();
+            // abort move should be successful, it only clears the transient state in this case though
+            cluster.get(2).nodetoolResult("abortmove").asserts().success();
+            assertNormal(cluster.get(2));
+            cluster.get(2).nodetoolResult("move", "1234").asserts().success();
+            assertNormal(cluster.get(2));
+        }
+    }
+
+    @Test
+    public void lostDecomCommitResponseTest() throws IOException
+    {
+        try (Cluster cluster = init(builder().withNodes(2)
+                                             .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP).set("cms_await_timeout", "1s").set("cms_default_max_retries", "5"))
+                                             .start()))
+        {
+            cluster.filters().verbs(Verb.TCM_COMMIT_RSP.id).from(1).to(2).drop();
+            cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure();
+            assertDecomFailed(cluster.get(2));
+            cluster.get(2).nodetoolResult("abortdecommission").asserts().failure();
+            assertNormal(cluster.get(2));
+            cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure();
+            assertDecomFailed(cluster.get(2));
+            cluster.get(2).nodetoolResult("decommission").asserts().success();
+        }
+    }
+
+    @Test
+    public void lostDecomCommitRequestTest() throws IOException
+    {
+        try (Cluster cluster = init(builder().withNodes(2)
+                                             .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP).set("cms_await_timeout", "1s").set("cms_default_max_retries", "5"))
+                                             .start()))
+        {
+            cluster.filters().verbs(Verb.TCM_COMMIT_REQ.id).from(2).to(1).drop();
+            cluster.get(2).nodetoolResult("decommission", "--force").asserts().failure();
+            assertDecomFailed(cluster.get(2));
+            cluster.filters().reset();
+            cluster.get(2).nodetoolResult("abortdecommission").asserts().success();
+            assertNormal(cluster.get(2));
+            cluster.get(2).nodetoolResult("decommission", "--force").asserts().success();
+        }
+    }
+
+    private static void assertNormal(IInvokableInstance i)
+    {
+        assertOperationMode(i, StorageService.Mode.NORMAL);
+    }
+
+    private static void assertMoveFailed(IInvokableInstance i)
+    {
+        assertOperationMode(i, StorageService.Mode.MOVE_FAILED);
+    }
+
+    private static void assertDecomFailed(IInvokableInstance i)
+    {
+        assertOperationMode(i, StorageService.Mode.DECOMMISSION_FAILED);
+    }
+
+    private static void assertOperationMode(IInvokableInstance i, StorageService.Mode expectedMode)
+    {
+        String mode = i.callOnInstance(() -> StorageService.instance.operationMode().toString());
+        assertEquals(expectedMode.toString(), mode);
+    }
+}

test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradePreparedStatementsTest.java

@@ -0,0 +1,110 @@
+/*
+ * 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.upgrade;
+
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+
+import org.junit.Test;
+
+import org.apache.cassandra.cql3.QueryHandler;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.statements.SelectStatement;
+import org.apache.cassandra.cql3.statements.UpdateStatement;
+import org.apache.cassandra.distributed.Constants;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.IUpgradeableInstance;
+
+public class ClusterMetadataUpgradePreparedStatementsTest extends UpgradeTestBase
+{
+    @Test
+    public void simpleUpgradeTest() throws Throwable
+    {
+        new UpgradeTestBase.TestCase()
+        .nodes(3)
+        .nodesToUpgrade(1, 2, 3)
+        .withConfig((cfg) -> cfg.with(Feature.NETWORK, Feature.GOSSIP, Feature.NATIVE_PROTOCOL)
+                                              .set(Constants.KEY_DTEST_FULL_STARTUP, true))
+        .upgradesToCurrentFrom(v50)
+        .setup((cluster) -> {
+            cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"));
+        })
+        .runAfterClusterUpgrade((cluster) -> {
+            String node1Address = cluster.get(1).config().broadcastAddress().getHostString();
+            com.datastax.driver.core.Cluster.Builder builder = com.datastax.driver.core.Cluster.builder()
+                                                                                               .addContactPoint(node1Address);
+            try (com.datastax.driver.core.Cluster c = builder.build();
+                 Session session = c.connect())
+            {
+                PreparedStatement ps = session.prepare(withKeyspace("insert into %s.tbl (pk, ck, v) values (?, ?, ?)"));
+                PreparedStatement ps2 = session.prepare(withKeyspace("select pk, ck, v from %s.tbl where pk = ?"));
+                for (int i = 0; i < 10; i++)
+                {
+                    session.execute(ps.bind(i, 2, 3));
+                    session.execute(ps2.bind(i));
+                }
+                assertPSCache(cluster.get(1), false, 0);
+                cluster.get(1).nodetoolResult("cms", "initialize").asserts().success();
+                assertPSCache(cluster.get(1), false, 0);
+                for (int i = 0; i < 10; i++)
+                {
+                    session.execute(ps.bind(i, 3, 3));
+                    session.execute(ps2.bind(i));
+                }
+                session.execute(withKeyspace("alter table %s.tbl add x int"));
+
+                assertPSCacheEmpty(cluster.get(1));
+
+                for (int i = 0; i < 10; i++)
+                {
+                    session.execute(ps.bind(i, 3, 3));
+                    session.execute(ps2.bind(i));
+                }
+                assertPSCache(cluster.get(1), false, 4);
+            }
+        }).run();
+    }
+
+    private static void assertPSCacheEmpty(IUpgradeableInstance inst)
+    {
+        assertPSCache(inst, true, -1);
+    }
+
+    private static void assertPSCache(IUpgradeableInstance inst, boolean shouldBeEmpty, long expectedEpoch)
+    {
+        ((IInvokableInstance)inst).runOnInstance(() -> {
+            if (shouldBeEmpty != QueryProcessor.instance.getPreparedStatements().isEmpty())
+                throw new AssertionError("Prepared statements should not be empty after `cms initialize`");
+
+            for (QueryHandler.Prepared p : QueryProcessor.instance.getPreparedStatements().values())
+            {
+                long statementEpoch = -1;
+                if (p.statement instanceof SelectStatement)
+                    statementEpoch = ((SelectStatement)p.statement).table.epoch.getEpoch();
+                else if (p.statement instanceof UpdateStatement)
+                    statementEpoch = ((UpdateStatement)p.statement).metadata.epoch.getEpoch();
+
+                if (statementEpoch != expectedEpoch)
+                    throw new AssertionError(String.format("Statement %s has the wrong epoch: %d != %d ", p.statement, statementEpoch, expectedEpoch));
+            }
+        });
+    }
+
+}

test/microbench/org/apache/cassandra/test/microbench/DirectorySerializationBench.java

@@ -0,0 +1,126 @@
+/*
+ * 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.test.microbench;
+
+import java.io.IOException;
+import java.net.UnknownHostException;
+import java.util.Collections;
+import java.util.Random;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.distributed.api.TokenSupplier;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.RegistrationStatus;
+import org.apache.cassandra.tcm.membership.Location;
+import org.apache.cassandra.tcm.membership.NodeAddresses;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.ownership.PlacementProvider;
+import org.apache.cassandra.tcm.ownership.UniformRangePlacement;
+import org.apache.cassandra.tcm.transformations.UnsafeJoin;
+import org.apache.cassandra.utils.CassandraVersion;
+
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Fork(value = 1)
+@Warmup(iterations = 5, timeUnit = TimeUnit.MILLISECONDS, time = 5000)
+@Measurement(iterations = 5, timeUnit = TimeUnit.MILLISECONDS, time = 30000)
+public class DirectorySerializationBench
+{
+    static Random random = new Random(1);
+    static ClusterMetadata metadata;
+    static byte[] serialized;
+    @Setup(Level.Trial)
+    public void setup() throws IOException
+    {
+        DatabaseDescriptor.daemonInitialization();
+        int nodecount = 4000;
+        metadata = fakeMetadata(nodecount, 3, 3);
+        RegistrationStatus.instance.onRegistration();
+        DataOutputBuffer buf = new DataOutputBuffer(2_000_000);
+        ClusterMetadata.serializer.serialize(metadata, buf, NodeVersion.CURRENT_METADATA_VERSION);
+        serialized = buf.toByteArray();
+    }
+
+    @Benchmark
+    public void bench() throws IOException
+    {
+        ClusterMetadata.serializer.deserialize(new DataInputBuffer(serialized), NodeVersion.CURRENT_METADATA_VERSION);
+    }
+
+    public static ClusterMetadata fakeMetadata(int nodeCount, int dcCount, int rackCount) throws UnknownHostException
+    {
+        ClusterMetadata metadata = new ClusterMetadata(Murmur3Partitioner.instance);
+        TokenSupplier tokensupplier = TokenSupplier.evenlyDistributedTokens(nodeCount);
+        PlacementProvider placementProvider = new UniformRangePlacement();
+        NodeVersion nodeVersion = new NodeVersion(new CassandraVersion("6.0.0"), NodeVersion.CURRENT_METADATA_VERSION);
+        for (int i = 1; i < nodeCount; i++)
+        {
+            ClusterMetadata.Transformer transformer = metadata.transformer();
+            UUID uuid = UUID.randomUUID();
+            NodeAddresses addresses = addresses(uuid, i);
+            metadata = transformer.register(addresses, new Location("dc" + random.nextInt(dcCount), "rack"+random.nextInt(rackCount)), nodeVersion).build().metadata;
+            NodeId nodeId = metadata.directory.peerId(addresses.broadcastAddress);
+            metadata = new UnsafeJoin(nodeId, Collections.singleton(new Murmur3Partitioner.LongToken(tokensupplier.token(i))), placementProvider).execute(metadata).success().metadata;
+        }
+
+        return metadata;
+    }
+
+    static NodeAddresses addresses(UUID uuid, int idx) throws UnknownHostException
+    {
+        byte [] address = new byte [] {127, 0,
+                                       (byte) (((idx + 1) & 0x0000ff00) >> 8),
+                                       (byte) ((idx + 1) & 0x000000ff)};
+
+        InetAddressAndPort host = InetAddressAndPort.getByAddress(address);
+        return new NodeAddresses(uuid, host, host, host);
+    }
+
+    public static void main(String[] args) throws RunnerException, UnknownHostException
+    {
+        Options options = new OptionsBuilder()
+                          .include(DirectorySerializationBench.class.getSimpleName())
+                          .build();
+        new Runner(options).run();
+    }
+}

test/unit/org/apache/cassandra/db/compaction/CompactionStrategyManagerTest.java

@@ -214,8 +214,9 @@ public class CompactionStrategyManagerTest extends CassandraTestBase
         // inside the currentlyBackgroundUpgrading check - with max_concurrent_auto_upgrade_tasks = 1 this will make
         // sure that BackgroundCompactionCandidate#maybeRunUpgradeTask returns false until the latch has been counted down
         CountDownLatch latch = new CountDownLatch(1);
+        CountDownLatch inFindUpgradeSSTables = new CountDownLatch(1);
         AtomicInteger upgradeTaskCount = new AtomicInteger(0);
-        MockCFSForCSM mock = new MockCFSForCSM(cfs, latch, upgradeTaskCount);
+        MockCFSForCSM mock = new MockCFSForCSM(cfs, latch, inFindUpgradeSSTables, upgradeTaskCount);
 
         CompactionManager.BackgroundCompactionCandidate r = CompactionManager.instance.getBackgroundCompactionCandidate(mock);
         CompactionStrategyManager mgr = mock.getCompactionStrategyManager();
@@ -224,7 +225,7 @@ public class CompactionStrategyManagerTest extends CassandraTestBase
         // due to the currentlyBackgroundUpgrading count being >= max_concurrent_auto_upgrade_tasks
         Thread t = new Thread(() -> r.maybeRunUpgradeTask(mgr));
         t.start();
-        Thread.sleep(100); // let the thread start and grab the task
+        inFindUpgradeSSTables.await();
         assertEquals(1, CompactionManager.instance.currentlyBackgroundUpgrading.get());
         assertFalse(r.maybeRunUpgradeTask(mgr));
         assertFalse(r.maybeRunUpgradeTask(mgr));
@@ -246,8 +247,9 @@ public class CompactionStrategyManagerTest extends CassandraTestBase
         // inside the currentlyBackgroundUpgrading check - with max_concurrent_auto_upgrade_tasks = 1 this will make
         // sure that BackgroundCompactionCandidate#maybeRunUpgradeTask returns false until the latch has been counted down
         CountDownLatch latch = new CountDownLatch(1);
+        CountDownLatch inFindUpgradeSSTables = new CountDownLatch(2);
         AtomicInteger upgradeTaskCount = new AtomicInteger();
-        MockCFSForCSM mock = new MockCFSForCSM(cfs, latch, upgradeTaskCount);
+        MockCFSForCSM mock = new MockCFSForCSM(cfs, latch, inFindUpgradeSSTables, upgradeTaskCount);
 
         CompactionManager.BackgroundCompactionCandidate r = CompactionManager.instance.getBackgroundCompactionCandidate(mock);
         CompactionStrategyManager mgr = mock.getCompactionStrategyManager();
@@ -259,7 +261,7 @@ public class CompactionStrategyManagerTest extends CassandraTestBase
         t.start();
         Thread t2 = new Thread(() -> r.maybeRunUpgradeTask(mgr));
         t2.start();
-        Thread.sleep(100); // let the threads start and grab the task
+        inFindUpgradeSSTables.await();
         assertEquals(2, CompactionManager.instance.currentlyBackgroundUpgrading.get());
         assertFalse(r.maybeRunUpgradeTask(mgr));
         assertFalse(r.maybeRunUpgradeTask(mgr));
@@ -619,18 +621,20 @@ public class CompactionStrategyManagerTest extends CassandraTestBase
     private static class MockCFSForCSM extends ColumnFamilyStore
     {
         private final CountDownLatch latch;
+        private final CountDownLatch inFindUpgradeSSTables;
         private final AtomicInteger upgradeTaskCount;
 
-        private MockCFSForCSM(ColumnFamilyStore cfs, CountDownLatch latch, AtomicInteger upgradeTaskCount)
+        private MockCFSForCSM(ColumnFamilyStore cfs, CountDownLatch latch, CountDownLatch inFindUpgradeSSTables, AtomicInteger upgradeTaskCount)
         {
             super(cfs.keyspace, cfs.name, Util.newSeqGen(10), cfs.metadata.get(), cfs.getDirectories(), true, false);
             this.latch = latch;
+            this.inFindUpgradeSSTables = inFindUpgradeSSTables;
             this.upgradeTaskCount = upgradeTaskCount;
         }
         @Override
         public CompactionStrategyManager getCompactionStrategyManager()
         {
-            return new MockCSM(this, latch, upgradeTaskCount);
+            return new MockCSM(this, latch, inFindUpgradeSSTables, upgradeTaskCount);
         }
     }
 
@@ -638,11 +642,13 @@ public class CompactionStrategyManagerTest extends CassandraTestBase
     {
         private final CountDownLatch latch;
         private final AtomicInteger upgradeTaskCount;
+        private final CountDownLatch inFindUpgradeSSTables;
 
-        private MockCSM(ColumnFamilyStore cfs, CountDownLatch latch, AtomicInteger upgradeTaskCount)
+        private MockCSM(ColumnFamilyStore cfs, CountDownLatch latch, CountDownLatch inFindUpgradeSSTables, AtomicInteger upgradeTaskCount)
         {
             super(cfs);
             this.latch = latch;
+            this.inFindUpgradeSSTables = inFindUpgradeSSTables;
             this.upgradeTaskCount = upgradeTaskCount;
         }
 
@@ -651,6 +657,7 @@ public class CompactionStrategyManagerTest extends CassandraTestBase
         {
             try
             {
+                inFindUpgradeSSTables.countDown();
                 latch.await();
                 upgradeTaskCount.incrementAndGet();
             }