From 1b78bbdc3557de3ccd3acef2d0eac3e6df9df533 Mon Sep 17 00:00:00 2001
From: Benedict Elliott Smith <benedict@apache.org>
Date: Wed, 11 Jun 2025 18:44:05 +0100
Subject: [PATCH] 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                                           | 2 +-
 .../service/accord/AccordConfigurationService.java       | 9 +++++----
 .../apache/cassandra/service/accord/AccordService.java   | 2 +-
 .../service/accord/AccordConfigurationServiceTest.java   | 5 +++--
 .../service/accord/AccordSyncPropagatorTest.java         | 3 ++-
 .../apache/cassandra/service/accord/EpochSyncTest.java   | 3 ++-
 6 files changed, 14 insertions(+), 10 deletions(-)

diff --git a/modules/accord b/modules/accord
index bf85660dcc..c2ef0643ab 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit bf85660dccaece9fcb3dd319b6f82265d780378f
+Subproject commit c2ef0643ab671067d39e7ce688eea493e9ee018d
diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
index fade204b1f..bd06cb9764 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
+++ b/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
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index 4bf89c9b1e..6529db1683 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/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,
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java
index 77d6e687d2..94e07b9a9e 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java
+++ b/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);
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java
index 948dc80afb..adaff55578 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java
+++ b/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
diff --git a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java b/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
index cc05e86db4..fec84f5d63 100644
--- a/test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
+++ b/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