Fix:

 - Do not query local topology when deciding what keys to fetch to avoid TopologyRetiredException
 - Ensure we propagate information back to the requesting CommandStore by using the store id to ensure it is included
 - BurnTest topology fetching was broken by earlier patch
 - Topology callbacks were not being invoked as we were not calling .begin()
 - Topology mismatch failure during notAccept phase was not being reported due to CoordinatePreAccept already having isDone==true

patch by Benedict; reviewed by David Capwell for CASSANDRA-20711

Author: belliottsmith

accord-core/src/main/java/accord/coordinate/CheckShards.java

@@ -39,7 +39,7 @@
  */
 public abstract class CheckShards<U extends Participants<?>> extends ReadCoordinator<CheckStatusReply>
 {
-    final U route;
+    final U query;
 
     /**
      * The epoch we want to fetch data from remotely
@@ -54,17 +54,17 @@ public abstract class CheckShards<U extends Participants<?>> extends ReadCoordin
     protected boolean truncated;
 
     // srcEpoch is either txnId.epoch() or executeAt.epoch()
-    protected CheckShards(Node node, TxnId txnId, U route, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf)
+    protected CheckShards(Node node, TxnId txnId, U query, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf)
     {
-        this(node, txnId, route, txnId.epoch(), includeInfo, bumpBallot, previouslyKnownToBeInvalidIf);
+        this(node, txnId, query, txnId.epoch(), includeInfo, bumpBallot, previouslyKnownToBeInvalidIf);
         Invariants.require(txnId.isVisible());
     }
 
-    protected CheckShards(Node node, TxnId txnId, U route, long srcEpoch, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf)
+    protected CheckShards(Node node, TxnId txnId, U query, long srcEpoch, IncludeInfo includeInfo, @Nullable Ballot bumpBallot, Infer.InvalidIf previouslyKnownToBeInvalidIf)
     {
-        super(node, topologyFor(node, txnId, route, srcEpoch), txnId);
+        super(node, topologyFor(node, txnId, query, srcEpoch), txnId);
         this.sourceEpoch = srcEpoch;
-        this.route = route;
+        this.query = query;
         this.includeInfo = includeInfo;
         this.bumpBallot = bumpBallot;
         this.previouslyKnownToBeInvalidIf = previouslyKnownToBeInvalidIf;
@@ -79,7 +79,7 @@ private static Topologies topologyFor(Node node, TxnId txnId, Unseekables<?> con
     @Override
     protected void contact(Id id)
     {
-        Participants<?> unseekables = route.slice(topologies().computeRangesForNode(id));
+        Participants<?> unseekables = query.slice(topologies().computeRangesForNode(id));
         node.send(id, new CheckStatus(txnId, unseekables, sourceEpoch, includeInfo, bumpBallot), this);
     }
 
@@ -119,7 +119,7 @@ protected Action process(Id from, CheckStatusReply reply)
     @Override
     protected void finishOnExhaustion()
     {
-        if (merged != null && merged.map.hasFullyTruncated(route)) finishOnFailure(new Truncated(txnId, null), false);
+        if (merged != null && merged.map.hasFullyTruncated(query)) finishOnFailure(new Truncated(txnId, null), false);
         else super.finishOnExhaustion();
     }
 }

accord-core/src/main/java/accord/coordinate/CoordinatePreAccept.java

@@ -142,7 +142,7 @@ void onNewEpochTopologyMismatch(TopologyMismatch mismatch)
         proposeInvalidate(node, node.uniqueTimestamp(Ballot::fromValues), txnId, route.homeKey(), (outcome, failure) -> {
             if (failure != null)
                 mismatch.addSuppressed(failure);
-            setFailure(mismatch);
+            callback.accept(null, mismatch);
         });
     }
 

accord-core/src/main/java/accord/coordinate/FetchData.java

@@ -0,0 +1,151 @@
+/*
+ * 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 accord.coordinate;
+
+import java.util.function.BiConsumer;
+
+import accord.local.Commands;
+import accord.local.Node;
+import accord.local.SafeCommand;
+import accord.local.SafeCommandStore;
+import accord.primitives.Known;
+import accord.messages.CheckStatus.CheckStatusOk;
+import accord.messages.CheckStatus.IncludeInfo;
+import accord.primitives.WithQuorum;
+import accord.primitives.Participants;
+import accord.primitives.Route;
+import accord.primitives.TxnId;
+import accord.utils.MapReduceConsume;
+
+import static accord.coordinate.Infer.InvalidIf.NotKnownToBeInvalid;
+import static accord.coordinate.Infer.InvalidateAndCallback.locallyInvalidateAndCallback;
+import static accord.local.CommandStores.*;
+import static accord.primitives.Known.Nothing;
+import static accord.primitives.WithQuorum.HasQuorum;
+
+/**
+ * Find some Route for a txnId using some known participants
+ */
+public class FetchSomeRoute extends CheckShards<Participants<?>>
+{
+    final LatentStoreSelector reportTo;
+    final BiConsumer<Route<?>, Throwable> callback;
+    FetchSomeRoute(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Participants<?> contactable, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback)
+    {
+        super(node, txnId, contactable, IncludeInfo.Route, null, invalidIf);
+        this.reportTo = reportTo;
+        this.callback = callback;
+    }
+
+    public static void fetchSomeRoute(Node node, TxnId txnId, Infer.InvalidIf invalidIf, Participants<?> unseekables, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback)
+    {
+        if (!node.topology().hasEpoch(txnId.epoch()))
+        {
+            node.withEpochAtLeast(txnId.epoch(), callback, () -> fetchSomeRoute(node, txnId, invalidIf, unseekables, reportTo, callback));
+            return;
+        }
+
+        FetchSomeRoute fetchSomeRoute = new FetchSomeRoute(node, txnId, invalidIf, unseekables, reportTo, callback);
+        fetchSomeRoute.start();
+    }
+
+
+    public static void fetchSomeRoute(Node node, TxnId txnId, Participants<?> contactable, BiConsumer<Route<?>, Throwable> callback)
+    {
+        fetchSomeRoute(node, txnId, contactable, LatentStoreSelector.standard(), callback);
+    }
+
+    public static void fetchSomeRoute(Node node, TxnId txnId, Participants<?> contactable, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback)
+    {
+        fetchSomeRoute(node, txnId, NotKnownToBeInvalid, contactable, reportTo, callback);
+    }
+
+    @Override
+    protected boolean isSufficient(CheckStatusOk ok)
+    {
+        return ok.route != null;
+    }
+
+    @Override
+    protected void onDone(Success success, Throwable failure)
+    {
+        if (failure != null) callback.accept(null, failure);
+        else
+        {
+            final Route<?> route = merged == null ? null : merged.route;
+            if (route == null)
+            {
+                Known known = Nothing;
+                if (merged != null)
+                    known = merged.finish(query, query, query, success.withQuorum, previouslyKnownToBeInvalidIf).knownFor(txnId, query, query);
+                reportRouteNotFound(node, success.withQuorum, known, txnId, query, reportTo, callback);
+            }
+            else
+            {
+                StoreSelector selector = reportTo.refine(txnId, null, query);
+                node.mapReduceConsumeLocal(txnId, selector, new MapReduceConsume<>()
+                {
+                    @Override
+                    public void accept(Object result, Throwable failure)
+                    {
+                        callback.accept(route, null);
+                    }
+
+                    @Override
+                    public Object apply(SafeCommandStore safeStore)
+                    {
+                        SafeCommand safeCommand = safeStore.ifInitialised(txnId);
+                        if (safeCommand != null)
+                            Commands.updateRoute(safeStore, safeCommand, route);
+                        return null;
+                    }
+
+                    @Override
+                    public Object reduce(Object o1, Object o2)
+                    {
+                        return null;
+                    }
+                });
+            }
+        }
+    }
+
+    private static void reportRouteNotFound(Node node, WithQuorum withQuorum, Known found, TxnId txnId, Participants<?> participants, LatentStoreSelector reportTo, BiConsumer<Route<?>, Throwable> callback)
+    {
+        switch (found.outcome())
+        {
+            default: throw new AssertionError("Unknown outcome: " + found.outcome());
+            case Abort:
+                locallyInvalidateAndCallback(node, txnId, reportTo.refine(txnId, null, participants), participants, null, callback);
+                break;
+
+            case Unknown:
+                if (withQuorum == HasQuorum && found.canProposeInvalidation())
+                {
+                    Invalidate.invalidate(node, txnId, participants, false, reportTo, (outcome, throwable) -> callback.accept(null, throwable));
+                    break;
+                }
+            case Erased:
+            case WasApply:
+            case Apply:
+                callback.accept(null, null);
+        }
+    }
+
+}