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

modules/accord

@@ -1 +1 @@
-Subproject commit bf85660dccaece9fcb3dd319b6f82265d780378f
+Subproject commit c2ef0643ab671067d39e7ce688eea493e9ee018d

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

@@ -32,6 +32,7 @@ import javax.annotation.concurrent.GuardedBy;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Sets;
 
+import accord.api.Agent;
 import accord.impl.AbstractConfigurationService;
 import accord.local.Node;
 import accord.primitives.Ranges;
@@ -138,17 +139,17 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
         }
     }
 
-    public AccordConfigurationService(Node.Id node, MessageDelivery messagingService, IFailureDetector failureDetector, ScheduledExecutorPlus scheduledTasks)
+    public AccordConfigurationService(Node.Id node, Agent agent, MessageDelivery messagingService, IFailureDetector failureDetector, ScheduledExecutorPlus scheduledTasks)
     {
-        super(node);
+        super(node, agent);
         this.syncPropagator = new AccordSyncPropagator(localId, this, messagingService, failureDetector, scheduledTasks, this);
         this.watermarkCollector = new WatermarkCollector();
         listeners.add(watermarkCollector);
     }
 
-    public AccordConfigurationService(Node.Id node)
+    public AccordConfigurationService(Node.Id node, Agent agent)
     {
-        this(node, MessagingService.instance(), FailureDetector.instance, ScheduledExecutors.scheduledTasks);
+        this(node, agent, MessagingService.instance(), FailureDetector.instance, ScheduledExecutors.scheduledTasks);
     }
 
     @Override

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

@@ -300,7 +300,7 @@ public class AccordService implements IAccordService, Shutdownable
         this.scheduler = new AccordScheduler();
         this.dataStore = new AccordDataStore();
         this.journal = new AccordJournal(DatabaseDescriptor.getAccord().journal);
-        this.configService = new AccordConfigurationService(localId);
+        this.configService = new AccordConfigurationService(localId, agent);
         this.fastPathCoordinator = AccordFastPathCoordinator.create(localId, configService);
         this.messageSink = new AccordMessageSink(agent, configService, callbacks);
         this.node = new Node(localId,

test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java

@@ -62,6 +62,7 @@ import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.Tables;
+import org.apache.cassandra.service.accord.api.AccordAgent;
 import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.ValidatingClusterMetadataService;
@@ -171,7 +172,7 @@ public class AccordConfigurationServiceTest
         try
         {
             journal = initJournal();
-            AccordConfigurationService service = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), ScheduledExecutors.scheduledTasks);
+            AccordConfigurationService service = new AccordConfigurationService(ID1, new AccordAgent(), new Messaging(), new MockFailureDetector(), ScheduledExecutors.scheduledTasks);
             AccordJournal journal_ = journal;
             TestListener listener = new TestListener(service, true)
             {
@@ -200,7 +201,7 @@ public class AccordConfigurationServiceTest
             Topology topology3 = createTopology(cms);
             service.reportTopology(topology3);
 
-            AccordConfigurationService loaded = new AccordConfigurationService(ID1, new Messaging(), new MockFailureDetector(), ScheduledExecutors.scheduledTasks);
+            AccordConfigurationService loaded = new AccordConfigurationService(ID1, new AccordAgent(), new Messaging(), new MockFailureDetector(), ScheduledExecutors.scheduledTasks);
             loaded.updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1));
             listener = new AbstractConfigurationServiceTest.TestListener(loaded, true);
             loaded.registerListener(listener);

test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java

@@ -70,6 +70,7 @@ import org.apache.cassandra.net.ConnectionType;
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.net.MessageDelivery;
 import org.apache.cassandra.net.RequestCallback;
+import org.apache.cassandra.service.accord.api.AccordAgent;
 import org.apache.cassandra.tcm.ValidatingClusterMetadataService;
 import org.apache.cassandra.tcm.serialization.Version;
 import org.apache.cassandra.utils.AccordGenerators;
@@ -419,7 +420,7 @@ public class AccordSyncPropagatorTest
 
             private ConfigService(Node.Id node)
             {
-                super(node);
+                super(node, new AccordAgent());
             }
 
             @Override

test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java

@@ -87,6 +87,7 @@ import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.TableParams;
 import org.apache.cassandra.schema.Tables;
 import org.apache.cassandra.service.accord.AccordConfigurationService.EpochSnapshot;
+import org.apache.cassandra.service.accord.api.AccordAgent;
 import org.apache.cassandra.service.consensus.TransactionalMode;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.ClusterMetadataService;
@@ -682,7 +683,7 @@ public class EpochSyncTest
                 // TODO (review): Should there be a real scheduler here? Is it possible to adapt the Scheduler interface to scheduler used in this test?
                 TimeService time = TimeService.ofNonMonotonic(globalExecutor::currentTimeMillis, TimeUnit.MILLISECONDS);
                 this.topology = new TopologyManager(SizeOfIntersectionSorter.SUPPLIER, new TestAgent.RethrowAgent(), id, time, new DefaultTimeouts(time));
-                config = new AccordConfigurationService(node, messagingService, failureDetector, scheduler);
+                config = new AccordConfigurationService(node, new AccordAgent(), messagingService, failureDetector, scheduler);
                 config.registerListener(new ConfigurationService.Listener()
                 {
                     @Override