CEP-15: (C*) Implement TopologySorter to prioritise hosts based on DynamicSnitch and/or topology layout

patch by Blake Eggleston, David Capwell; reviewed by Blake Eggleston for CASSANDRA-18929

modules/accord

@@ -1 +1 @@
-Subproject commit 5ffe3d504bb5aa1ff1c2b96d817791e40f7ced0f
+Subproject commit d99ad84cc49a96299a9ae55183e38ee6f1aa3f47

src/java/org/apache/cassandra/locator/AbstractNetworkTopologySnitch.java

@@ -17,6 +17,10 @@
  */
 package org.apache.cassandra.locator;
 
+import java.util.Comparator;
+
+import org.apache.cassandra.utils.Sortable;
+
 /**
  * An endpoint snitch tells Cassandra information about network topology that it can use to route
  * requests more efficiently.
@@ -31,4 +35,16 @@ public abstract class AbstractNetworkTopologySnitch extends AbstractEndpointSnit
     {
         return proximity.compareEndpoints(address, r1, r2);
     }
+
+    @Override
+    public boolean supportCompareByEndpoint()
+    {
+        return proximity.supportCompareByEndpoint();
+    }
+
+    @Override
+    public <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+    {
+        return proximity.endpointComparator(address, addresses);
+    }
 }

src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java

@@ -42,6 +42,7 @@ import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.MBeanWrapper;
+import org.apache.cassandra.utils.Sortable;
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORE_DYNAMIC_SNITCH_SEVERITY;
 
@@ -170,10 +171,13 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers.
 
         // TODO: avoid copy
         replicas = delegate.sortedByProximity(address, replicas);
-        HashMap<InetAddressAndPort, Double> scores = this.scores; // Make sure the score don't change in the middle of the loop below
+        return shouldSortByScore(scores, replicas) ? sortedByProximityWithScore(address, replicas) : replicas;
-                                                           // (which wouldn't really matter here but its cleaner that way).
+    }
-        ArrayList<Double> subsnitchOrderedScores = new ArrayList<>(replicas.size());
+
-        for (Replica replica : replicas)
+    private <C extends Sortable<? extends Endpoint, ? extends C>> boolean shouldSortByScore(HashMap<InetAddressAndPort, Double> scores, C sortedReplicas)
+    {
+        ArrayList<Double> subsnitchOrderedScores = new ArrayList<>(sortedReplicas.size());
+        for (Endpoint replica : sortedReplicas)
         {
             Double score = scores.get(replica.endpoint());
             if (score == null)
@@ -193,12 +197,10 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers.
         for (Double subsnitchScore : subsnitchOrderedScores)
         {
             if (subsnitchScore > (sortedScoreIterator.next() * badnessThreshold))
-            {
+                return true;
-                return sortedByProximityWithScore(address, replicas);
-            }
         }
 
-        return replicas;
+        return false;
     }
 
     private static double defaultStore(InetAddressAndPort target)
@@ -208,6 +210,11 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers.
 
     // Compare endpoints given an immutable snapshot of the scores
     private int compareEndpoints(InetAddressAndPort target, Replica a1, Replica a2, Map<InetAddressAndPort, Double> scores)
+    {
+        return compareEndpoints(a1, a2, scores, (a, b) -> delegate.compareEndpoints(target, a, b));
+    }
+
+    private <T extends Endpoint> int compareEndpoints(T a1, T a2, Map<InetAddressAndPort, Double> scores, Comparator<T> subCompare)
     {
         Double scored1 = scores.get(a1.endpoint());
         Double scored2 = scores.get(a2.endpoint());
@@ -223,7 +230,7 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers.
         }
 
         if (scored1.equals(scored2))
-            return delegate.compareEndpoints(target, a1, a2);
+            return subCompare.compare(a1, a2);
         if (scored1 < scored2)
             return -1;
         else
@@ -409,4 +416,26 @@ public class DynamicEndpointSnitch implements NodeProximity, LatencySubscribers.
         }
         return maxScore;
     }
+
+    @Override
+    public boolean supportCompareByEndpoint()
+    {
+        return delegate.supportCompareByEndpoint();
+    }
+
+    @Override
+    public <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+    {
+        if (!delegate.supportCompareByEndpoint())
+            throw new UnsupportedOperationException();
+        assert address.equals(FBUtilities.getBroadcastAddressAndPort()); // we only know about ourself
+        Comparator<Endpoint> compare = delegate.endpointComparator(address, addresses);
+        if (addresses.size() < 2)
+            return compare;
+        HashMap<InetAddressAndPort, Double> scores = this.scores;
+        Comparator<Endpoint> compareWithScore = (r1, r2) -> compareEndpoints(r1, r2, scores, compare);
+        return dynamicBadnessThreshold == 0 || shouldSortByScore(scores, addresses.sorted(compare)) ?
+               compareWithScore :
+               compare;
+    }
 }

src/java/org/apache/cassandra/locator/Endpoint.java

@@ -0,0 +1,24 @@
+/*
+ * 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.locator;
+
+public interface Endpoint
+{
+    InetAddressAndPort endpoint();
+}

src/java/org/apache/cassandra/locator/IEndpointSnitch.java

@@ -18,9 +18,11 @@
 package org.apache.cassandra.locator;
 
 import java.net.InetSocketAddress;
+import java.util.Comparator;
 import java.util.Set;
 
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Sortable;
 
 /**
  * This interface helps determine location of node in the datacenter relative to another node.
@@ -101,5 +103,14 @@ public interface IEndpointSnitch
     {
         return false;
     }
-}
 
+    default boolean supportCompareByEndpoint()
+    {
+        return false;
+    }
+
+    default <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+    {
+        throw new UnsupportedOperationException();
+    }
+}

src/java/org/apache/cassandra/locator/NetworkTopologyProximity.java

@@ -20,10 +20,18 @@ package org.apache.cassandra.locator;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.tcm.membership.Location;
+import org.apache.cassandra.utils.Sortable;
+
+import java.util.Comparator;
 
 public class NetworkTopologyProximity extends BaseProximity
 {
     public int compareEndpoints(InetAddressAndPort address, Replica r1, Replica r2)
+    {
+        return compareByEndpoints(address, r1, r2);
+    }
+
+    public int compareByEndpoints(InetAddressAndPort address, Endpoint r1, Endpoint r2)
     {
         InetAddressAndPort a1 = r1.endpoint();
         InetAddressAndPort a2 = r2.endpoint();
@@ -48,4 +56,18 @@ public class NetworkTopologyProximity extends BaseProximity
             return 1;
         return 0;
     }
+
+    @Override
+    public boolean supportCompareByEndpoint()
+    {
+        return true;
+    }
+
+    @Override
+    public <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+    {
+        if (!supportCompareByEndpoint())
+            throw new UnsupportedOperationException();
+        return (a, b) -> compareByEndpoints(address, a, b);
+    }
 }

src/java/org/apache/cassandra/locator/NoOpProximity.java

@@ -18,6 +18,10 @@
 
 package org.apache.cassandra.locator;
 
+import org.apache.cassandra.utils.Sortable;
+
+import java.util.Comparator;
+
 public class NoOpProximity extends BaseProximity
 {
     @Override
@@ -34,4 +38,23 @@ public class NoOpProximity extends BaseProximity
         // Collections.sort is guaranteed to be stable)
         return 0;
     }
+
+    @Override
+    public boolean supportCompareByEndpoint()
+    {
+        return true;
+    }
+
+    @Override
+    public <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+    {
+        return this::compareByEndpoint;
+    }
+
+    private int compareByEndpoint(Endpoint a, Endpoint b)
+    {
+        // Making all endpoints equal ensures we won't change the original ordering (since
+        // Collections.sort is guaranteed to be stable)
+        return 0;
+    }
 }

src/java/org/apache/cassandra/locator/NodeProximity.java

@@ -18,6 +18,10 @@
 
 package org.apache.cassandra.locator;
 
+import org.apache.cassandra.utils.Sortable;
+
+import java.util.Comparator;
+
 public interface NodeProximity
 {
     /**
@@ -35,4 +39,14 @@ public interface NodeProximity
      * to be faster than 2 sequential queries, one against l1 followed by one against l2.
      */
     public boolean isWorthMergingForRangeQuery(ReplicaCollection<?> merged, ReplicaCollection<?> l1, ReplicaCollection<?> l2);
+
+    default boolean supportCompareByEndpoint()
+    {
+        return false;
+    }
+
+    default <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+    {
+        throw new UnsupportedOperationException();
+    }
 }

src/java/org/apache/cassandra/locator/Replica.java

@@ -51,7 +51,7 @@ import static org.apache.cassandra.dht.AbstractBounds.tokenSerializer;
  * and such and what the result is WRT to transientness. Definitely avoid creating fake Replicas with misinformation
  * about endpoints, ranges, or transientness.
  */
-public final class Replica implements Comparable<Replica>
+public final class Replica implements Comparable<Replica>, Endpoint
 {
     public static final IPartitionerDependentSerializer<Replica> serializer = new Serializer();
 
@@ -105,6 +105,7 @@ public final class Replica implements Comparable<Replica>
         return (full ? "Full" : "Transient") + '(' + endpoint() + ',' + range + ')';
     }
 
+    @Override
     public final InetAddressAndPort endpoint()
     {
         return endpoint;

src/java/org/apache/cassandra/locator/ReplicaCollection.java

@@ -24,11 +24,13 @@ import java.util.Set;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 
+import org.apache.cassandra.utils.Sortable;
+
 /**
  * A collection like class for Replica objects. Represents both a well defined order on the contained Replica objects,
  * and efficient methods for accessing the contained Replicas, directly and as a projection onto their endpoints and ranges.
  */
-public interface ReplicaCollection<C extends ReplicaCollection<C>> extends Iterable<Replica>
+public interface ReplicaCollection<C extends ReplicaCollection<C>> extends Sortable<Replica, C>
 {
     /**
      * @return a Set of the endpoints of the contained Replicas.

src/java/org/apache/cassandra/locator/SimpleSnitch.java

@@ -17,6 +17,10 @@
  */
 package org.apache.cassandra.locator;
 
+import java.util.Comparator;
+
+import org.apache.cassandra.utils.Sortable;
+
 /**
  * A simple endpoint snitch implementation that treats Strategy order as proximity,
  * allowing non-read-repaired reads to prefer a single endpoint, which improves
@@ -58,4 +62,16 @@ public class SimpleSnitch extends AbstractEndpointSnitch
     {
         return sorter.isWorthMergingForRangeQuery(merged, l1, l2);
     }
+
+    @Override
+    public boolean supportCompareByEndpoint()
+    {
+        return sorter.supportCompareByEndpoint();
+    }
+
+    @Override
+    public <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+    {
+        return sorter.endpointComparator(address, addresses);
+    }
 }

src/java/org/apache/cassandra/locator/SnitchAdapter.java

@@ -18,12 +18,14 @@
 
 package org.apache.cassandra.locator;
 
+import java.util.Comparator;
 import java.util.HashSet;
 import java.util.Set;
 
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.membership.Location;
+import org.apache.cassandra.utils.Sortable;
 
 public class SnitchAdapter implements InitialLocationProvider, NodeProximity, NodeAddressConfig
 {
@@ -81,4 +83,16 @@ public class SnitchAdapter implements InitialLocationProvider, NodeProximity, No
     {
         return snitch.preferLocalConnections();
     }
+
+    @Override
+    public boolean supportCompareByEndpoint()
+    {
+        return snitch.supportCompareByEndpoint();
+    }
+
+    @Override
+    public <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+    {
+        return snitch.endpointComparator(address, addresses);
+    }
 }

src/java/org/apache/cassandra/service/accord/AccordJournal.java

@@ -613,7 +613,7 @@ public class AccordJournal implements Shutdownable
     static
     {
         // make noise early if we forget to update our version mappings
-        Invariants.checkState(MessagingService.current_version == MessagingService.VERSION_50);
+        Invariants.checkState(MessagingService.current_version == MessagingService.VERSION_50, "Expected current version to be %d but given %d", MessagingService.VERSION_50, MessagingService.current_version);
     }
 
     private static int msVersion(int version)

src/java/org/apache/cassandra/service/accord/AccordService.java

@@ -71,6 +71,8 @@ import org.apache.cassandra.service.accord.AccordSyncPropagator.Notification;
 import org.apache.cassandra.service.accord.api.AccordAgent;
 import org.apache.cassandra.service.accord.api.AccordRoutingKey.KeyspaceSplitter;
 import org.apache.cassandra.service.accord.api.AccordScheduler;
+import org.apache.cassandra.service.accord.api.AccordTopologySorter;
+import org.apache.cassandra.service.accord.api.CompositeTopologySorter;
 import org.apache.cassandra.service.accord.exceptions.ReadPreemptedException;
 import org.apache.cassandra.service.accord.exceptions.WritePreemptedException;
 import org.apache.cassandra.service.accord.txn.TxnData;
@@ -138,7 +140,17 @@ public class AccordService implements IAccordService, Shutdownable
         }
 
         @Override
-        public void startup() {}
+        public void startup()
+        {
+            try
+            {
+                AccordTopologySorter.checkSnitchSupported(DatabaseDescriptor.getNodeProximity());
+            }
+            catch (Throwable t)
+            {
+                logger.warn("Current snitch  is not compatable with Accord, make sure to fix the snitch before enabling Accord; {}", t.toString());
+            }
+        }
 
         @Override
         public void shutdownAndWait(long timeout, TimeUnit unit) { }
@@ -241,7 +253,8 @@ public class AccordService implements IAccordService, Shutdownable
                              agent,
                              new DefaultRandom(),
                              scheduler,
-                             SizeOfIntersectionSorter.SUPPLIER,
+                             CompositeTopologySorter.create(SizeOfIntersectionSorter.SUPPLIER,
+                                                            new AccordTopologySorter.Supplier(configService, DatabaseDescriptor.getNodeProximity())),
                              SimpleProgressLog::new,
                              AccordCommandStores.factory(journal),
                              configuration);

src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java

@@ -35,6 +35,7 @@ import accord.topology.Topology;
 import accord.utils.Invariants;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.locator.EndpointsForRange;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.schema.DistributedSchema;
 import org.apache.cassandra.schema.KeyspaceMetadata;
@@ -47,7 +48,6 @@ import org.apache.cassandra.tcm.membership.Directory;
 import org.apache.cassandra.tcm.membership.NodeId;
 import org.apache.cassandra.tcm.ownership.DataPlacement;
 import org.apache.cassandra.tcm.ownership.DataPlacements;
-import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
 
 public class AccordTopologyUtils
 {
@@ -56,7 +56,7 @@ public class AccordTopologyUtils
         return new Node.Id(nodeId.id());
     }
 
-    private static Shard createShard(TokenRange range, Directory directory, VersionedEndpoints.ForRange reads, VersionedEndpoints.ForRange writes)
+    private static Shard createShard(TokenRange range, Directory directory, EndpointsForRange reads, EndpointsForRange writes)
     {
         Function<InetAddressAndPort, Node.Id> endpointMapper = e -> {
             NodeId tcmId = directory.peerId(e);
@@ -106,8 +106,10 @@ public class AccordTopologyUtils
         List<Shard> shards = new ArrayList<>(ranges.size());
         for (Range<Token> range : ranges)
         {
-            VersionedEndpoints.ForRange reads = placement.reads.forRange(range);
+            // TODO (consider, low priority): flesh out how Accord and Transient Replicas work together
-            VersionedEndpoints.ForRange writes = placement.reads.forRange(range);
+            // Accord needs to be able to read the full data from a single replica, but with transient ones they may only have a hash.
+            EndpointsForRange reads = placement.reads.forRange(range).get().filter(r -> r.isFull());
+            EndpointsForRange writes = placement.writes.forRange(range).get().filter(r -> r.isFull());
 
             // TCM doesn't create wrap around ranges
             Invariants.checkArgument(!range.isWrapAround() || range.right.equals(range.right.minValue()),

src/java/org/apache/cassandra/service/accord/api/AccordTopologySorter.java

@@ -0,0 +1,130 @@
+/*
+ * 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.service.accord.api;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Set;
+
+import accord.api.TopologySorter;
+import accord.local.Node;
+import accord.topology.ShardSelection;
+import accord.topology.Topologies;
+import accord.topology.Topology;
+import org.apache.cassandra.locator.*;
+import org.apache.cassandra.service.accord.AccordEndpointMapper;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Sortable;
+
+public class AccordTopologySorter implements TopologySorter
+{
+    public static class Supplier implements TopologySorter.Supplier
+    {
+        private final AccordEndpointMapper mapper;
+        private final NodeProximity proximity;
+
+        public Supplier(AccordEndpointMapper mapper, NodeProximity proximity)
+        {
+            checkSnitchSupported(proximity);
+            this.mapper = mapper;
+            this.proximity = proximity;
+        }
+
+        @Override
+        public TopologySorter get(Topology topologies)
+        {
+            return create(topologies.nodes());
+        }
+
+        @Override
+        public TopologySorter get(Topologies topologies)
+        {
+            return create(topologies.nodes());
+        }
+
+        private AccordTopologySorter create(Set<Node.Id> nodes)
+        {
+            SortableEndpoints endpoints = SortableEndpoints.from(nodes, mapper);
+            Comparator<Endpoint> comparator = proximity.endpointComparator(FBUtilities.getBroadcastAddressAndPort(), endpoints);
+            return new AccordTopologySorter(mapper, comparator);
+        }
+    }
+
+    private final AccordEndpointMapper mapper;
+
+    private final Comparator<Endpoint> comparator;
+    private AccordTopologySorter(AccordEndpointMapper mapper, Comparator<Endpoint> comparator)
+    {
+        this.mapper = mapper;
+        this.comparator = comparator;
+    }
+
+    public static void checkSnitchSupported(NodeProximity proximity)
+    {
+        if (!proximity.supportCompareByEndpoint())
+        {
+            if (proximity instanceof DynamicEndpointSnitch)
+                proximity = ((DynamicEndpointSnitch) proximity).delegate;
+            throw new IllegalArgumentException("Unsupported snitch " + proximity.getClass() + "; supportCompareByEndpoint returned false");
+        }
+    }
+
+    @Override
+    public int compare(Node.Id node1, Node.Id node2, ShardSelection shards)
+    {
+        return comparator.compare(() -> mapper.mappedEndpoint(node1), () -> mapper.mappedEndpoint(node2));
+    }
+
+    private static class EndpointTuple implements Endpoint
+    {
+        final InetAddressAndPort endpoint;
+
+        private EndpointTuple(InetAddressAndPort endpoint)
+        {
+            this.endpoint = endpoint;
+        }
+
+        @Override
+        public InetAddressAndPort endpoint()
+        {
+            return endpoint;
+        }
+    }
+
+    private static class SortableEndpoints extends ArrayList<EndpointTuple> implements Sortable<EndpointTuple, SortableEndpoints>
+    {
+        public SortableEndpoints(int initialCapacity)
+        {
+            super(initialCapacity);
+        }
+
+        public SortableEndpoints sorted(Comparator<? super EndpointTuple> comparator)
+        {
+            sort(comparator);
+            return this;
+        }
+
+        static SortableEndpoints from(Set<Node.Id> nodes, AccordEndpointMapper mapper)
+        {
+            SortableEndpoints result = new SortableEndpoints(nodes.size());
+            nodes.forEach(id -> result.add(new EndpointTuple(mapper.mappedEndpoint(id))));
+            return result;
+        }
+    }
+}

src/java/org/apache/cassandra/service/accord/api/CompositeTopologySorter.java

@@ -0,0 +1,84 @@
+/*
+ * 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.service.accord.api;
+
+import accord.api.TopologySorter;
+import accord.local.Node;
+import accord.topology.ShardSelection;
+import accord.topology.Topologies;
+import accord.topology.Topology;
+
+public class CompositeTopologySorter implements TopologySorter
+{
+    public static class Supplier implements TopologySorter.Supplier
+    {
+        private final TopologySorter.Supplier[] delegates;
+
+        private Supplier(TopologySorter.Supplier[] delegates)
+        {
+            this.delegates = delegates;
+        }
+
+        @Override
+        public TopologySorter get(Topology topologies)
+        {
+            TopologySorter[] sorters = new TopologySorter[delegates.length];
+            for (int i = 0; i < sorters.length; i++)
+                sorters[i] = delegates[i].get(topologies);
+            return new CompositeTopologySorter(sorters);
+        }
+
+        @Override
+        public TopologySorter get(Topologies topologies)
+        {
+            TopologySorter[] sorters = new TopologySorter[delegates.length];
+            for (int i = 0; i < sorters.length; i++)
+                sorters[i] = delegates[i].get(topologies);
+            return new CompositeTopologySorter(sorters);
+        }
+    }
+
+    private final TopologySorter[] delegates;
+
+    private CompositeTopologySorter(TopologySorter[] delegates)
+    {
+        this.delegates = delegates;
+    }
+
+    public static TopologySorter.Supplier create(TopologySorter.Supplier... delegates)
+    {
+        switch (delegates.length)
+        {
+            case 0:     throw new IllegalArgumentException("Can not create an empty sorter");
+            case 1:     return delegates[0];
+            default:    return new CompositeTopologySorter.Supplier(delegates);
+        }
+    }
+
+    @Override
+    public int compare(Node.Id node1, Node.Id node2, ShardSelection shards)
+    {
+        for (int i = 0; i < delegates.length; i++)
+        {
+            int rc = delegates[i].compare(node1, node2, shards);
+            if (rc != 0) return rc;
+        }
+        return 0;
+    }
+}

src/java/org/apache/cassandra/utils/Sortable.java

@@ -0,0 +1,28 @@
+/*
+ * 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.utils;
+
+import java.util.Comparator;
+
+public interface Sortable<T, S extends Sortable<T, S>> extends Iterable<T>
+{
+    int size();
+
+    S sorted(Comparator<? super T> comparator);
+}

test/simulator/main/org/apache/cassandra/simulator/paxos/AccordClusterSimulation.java

@@ -63,7 +63,7 @@ class AccordClusterSimulation extends ClusterSimulation<PaxosSimulation> impleme
     AccordClusterSimulation(RandomSource random, long seed, int uniqueNum, Builder builder) throws IOException
     {
         super(random, seed, uniqueNum, builder,
-              config -> {},
+              config -> config.set("storage_compatibility_mode", "NONE"),
               (simulated, schedulers, cluster, options) -> {
                   int[] primaryKeys = primaryKeys(seed, builder.primaryKeyCount());
                   KindOfSequence.Period jitter = RandomSource.Choices.uniform(KindOfSequence.values()).choose(random)

test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosClusterSimulation.java

@@ -79,6 +79,7 @@ class PaxosClusterSimulation extends ClusterSimulation<PaxosSimulation> implemen
                               .set("paxos_cache_size", (builder.stateCache != null ? builder.stateCache : random.uniformFloat() < 0.5) ? null : "0MiB")
                               .set("paxos_state_purging", "repaired")
                               .set("paxos_on_linearizability_violations", "log")
+                              .set("storage_compatibility_mode", "NONE")
         ,
               (simulated, schedulers, cluster, options) -> {
                   int[] primaryKeys = primaryKeys(seed, builder.primaryKeyCount());

test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedSnitch.java

@@ -28,16 +28,49 @@ import java.util.stream.IntStream;
 
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.IInstanceConfig;
-import org.apache.cassandra.locator.IEndpointSnitch;
+import org.apache.cassandra.locator.*;
-import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.locator.Replica;
-import org.apache.cassandra.locator.ReplicaCollection;
 import org.apache.cassandra.simulator.cluster.NodeLookup;
+import org.apache.cassandra.utils.Sortable;
 
 public class SimulatedSnitch extends NodeLookup
 {
+    private static class SimulatedProximity implements NodeProximity
+    {
+        @Override
+        public <C extends ReplicaCollection<? extends C>> C sortedByProximity(InetAddressAndPort address, C addresses)
+        {
+            return addresses.sorted(Comparator.comparingInt(SimulatedSnitch::asInt));
+        }
+
+        @Override
+        public int compareEndpoints(InetAddressAndPort target, Replica r1, Replica r2)
+        {
+            return Comparator.comparingInt(SimulatedSnitch::asInt).compare(r1, r2);
+        }
+
+        @Override
+        public boolean isWorthMergingForRangeQuery(ReplicaCollection<?> merged, ReplicaCollection<?> l1, ReplicaCollection<?> l2)
+        {
+            return false;
+        }
+
+        @Override
+        public boolean supportCompareByEndpoint()
+        {
+            return true;
+        }
+
+        @Override
+        public <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+        {
+            return Comparator.comparingInt(SimulatedSnitch::asInt);
+        }
+    }
+    
     public static class Instance implements IEndpointSnitch
     {
+        private final NodeProximity proximity = new SimulatedProximity();
+
         private static volatile Function<InetSocketAddress, String> LOOKUP_DC;
 
         public String getRack(InetAddressAndPort endpoint)
@@ -52,12 +85,12 @@ public class SimulatedSnitch extends NodeLookup
 
         public <C extends ReplicaCollection<? extends C>> C sortedByProximity(InetAddressAndPort address, C addresses)
         {
-            return addresses.sorted(Comparator.comparingInt(SimulatedSnitch::asInt));
+            return proximity.sortedByProximity(address, addresses);
         }
 
         public int compareEndpoints(InetAddressAndPort target, Replica r1, Replica r2)
         {
-            return Comparator.comparingInt(SimulatedSnitch::asInt).compare(r1, r2);
+            return proximity.compareEndpoints(target, r1, r2);
         }
 
         public void gossiperStarting()
@@ -66,7 +99,7 @@ public class SimulatedSnitch extends NodeLookup
 
         public boolean isWorthMergingForRangeQuery(ReplicaCollection<?> merged, ReplicaCollection<?> l1, ReplicaCollection<?> l2)
         {
-            return false;
+            return proximity.isWorthMergingForRangeQuery(merged, l1, l2);
         }
 
         public static void setup(Function<InetSocketAddress, String> lookupDc)
@@ -127,7 +160,7 @@ public class SimulatedSnitch extends NodeLookup
         return Arrays.asList(nameOfDcs);
     }
 
-    private static int asInt(Replica address)
+    private static int asInt(Endpoint address)
     {
         byte[] bytes = address.endpoint().addressBytes;
         return bytes[0] | (bytes[1] << 8) | (bytes[2] << 16) | (bytes[3] << 24);

test/unit/org/apache/cassandra/ServerTestUtils.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.net.UnknownHostException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -44,6 +45,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.format.big.BigTableReader;
 import org.apache.cassandra.io.sstable.indexsummary.IndexSummarySupport;
 import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.locator.Endpoint;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.locator.BaseProximity;
@@ -70,6 +72,7 @@ import org.apache.cassandra.tcm.transformations.Register;
 import org.apache.cassandra.tcm.transformations.UnsafeJoin;
 import org.apache.cassandra.tcm.transformations.cms.Initialize;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.Sortable;
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION;
 
@@ -108,6 +111,18 @@ public final class ServerTestUtils
             {
                 return 0;
             }
+
+            @Override
+            public boolean supportCompareByEndpoint()
+            {
+                return true;
+            }
+
+            @Override
+            public <C extends Sortable<? extends Endpoint, ? extends C>> Comparator<Endpoint> endpointComparator(InetAddressAndPort address, C addresses)
+            {
+                return (a, b) -> 0;
+            }
         });
     }
 

test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java

@@ -272,6 +272,7 @@ public class DatabaseDescriptorRefTest
     "org.apache.cassandra.io.util.PathUtils$IOToLongFunction",
     "org.apache.cassandra.io.util.RebufferingInputStream",
     "org.apache.cassandra.io.util.SpinningDiskOptimizationStrategy",
+    "org.apache.cassandra.locator.Endpoint",
     "org.apache.cassandra.locator.IEndpointSnitch",
     "org.apache.cassandra.locator.InetAddressAndPort",
     "org.apache.cassandra.locator.Locator",

test/unit/org/apache/cassandra/locator/NodeProximityEndpointCompareTest.java

@@ -0,0 +1,72 @@
+/*
+ * 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.locator;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Modifier;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import org.reflections.Reflections;
+import org.reflections.scanners.Scanners;
+import org.reflections.util.ConfigurationBuilder;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+
+public class NodeProximityEndpointCompareTest
+{
+    static
+    {
+        DatabaseDescriptor.clientInitialization();
+    }
+
+    @Test
+    public void allSupportEndpoint() throws InvocationTargetException, InstantiationException, IllegalAccessException
+    {
+        Reflections reflections = new Reflections(new ConfigurationBuilder()
+                                                  .forPackage("org.apache.cassandra")
+                                                  .setScanners(Scanners.SubTypes)
+                                                  .setExpandSuperTypes(true));
+
+        for (Class<? extends NodeProximity> klass : reflections.getSubTypesOf(NodeProximity.class))
+        {
+            if (Modifier.isAbstract(klass.getModifiers())
+                || Modifier.isPrivate(klass.getModifiers()) // private can not be created normally, so these are scoped to tests and can be ignored
+                || klass.isAnonymousClass())
+                continue;
+            Constructor<? extends NodeProximity> declaredConstructor;
+            try
+            {
+                declaredConstructor = klass.getDeclaredConstructor();
+            }
+            catch (NoSuchMethodException e)
+            {
+                // DynamicEndpointSnitch or test snitch... we can not create this normally
+                continue;
+            }
+            if (Modifier.isPrivate(declaredConstructor.getModifiers()))
+                continue;
+            NodeProximity proximity = declaredConstructor.newInstance();
+            Assertions.assertThat(proximity.supportCompareByEndpoint())
+                      .describedAs("Snitch %s does not support compare by endpoint!", proximity.getClass())
+                      .isTrue();
+        }
+    }
+}