Merge branch 'cassandra-6.0' into trunk

Author: belliottsmith

modules/accord

@@ -1263,7 +1263,6 @@ public void truncate()
         @Override
         public void collect(PartitionsCollector collector)
         {
-            int nodeId = AccordService.unsafeInstance().nodeId().id;
             tracing().forEach(id -> true, (txnId, events) -> {
                 events.forEach(e -> {
                     if (e.messages().isEmpty())

src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java

@@ -309,6 +309,7 @@ private enum State { INIT, STARTING, STARTED, STOPPED, SHUTTING_DOWN, SHUTDOWN }
     //  tests can specify a DelegatingService if they want to override
     private static IAccordService instance;
     private static IAccordService unsafeInstance;
+    private static volatile Node.Id nodeId;
     private static volatile IAccordService requestInstance;
     private static volatile IAccordService replyInstance;
 
@@ -378,6 +379,7 @@ public synchronized static void localStartup(NodeId tcmId)
         }
         else
         {
+            nodeId = new Node.Id(tcmId.id());
             AccordService as = new AccordService(tcmIdToAccord(tcmId));
             unsafeInstance = replyInstance = as;
             as.localStartup();
@@ -438,6 +440,11 @@ public static IAccordService unsafeInstance()
         return i;
     }
 
+    public static Node.Id nodeId()
+    {
+        return nodeId;
+    }
+
     public static boolean started()
     {
         if (!DatabaseDescriptor.getAccordTransactionsEnabled())
@@ -513,7 +520,7 @@ public static void applyProtocolModifiers(AccordConfig config)
         if (config.send_stable != null)
             ProtocolModifiers.Configure.setSendStableMessages(config.send_stable);
         if (config.send_minimal != null)
-            ProtocolModifiers.Configure.setSendMinimalCommits(config.send_minimal);
+            ProtocolModifiers.Configure.setSendMinimal(config.send_minimal);
     }
 
     @Override
@@ -1306,11 +1313,6 @@ public AccordScheduler scheduler()
         return scheduler;
     }
 
-    public Id nodeId()
-    {
-        return node.id();
-    }
-
     @Override
     public long minEpoch()
     {

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

@@ -181,8 +181,6 @@ public AccordCompactionInfos(DurableBefore durableBefore, long minEpoch, AccordC
 
     Agent agent();
 
-    Id nodeId();
-
     long minEpoch();
 
     void awaitDone(TableId id, long epoch);
@@ -338,12 +336,6 @@ public Agent agent()
             return null;
         }
 
-        @Override
-        public Id nodeId()
-        {
-            throw new UnsupportedOperationException();
-        }
-
         @Override
         public long minEpoch()
         {
@@ -548,12 +540,6 @@ public Agent agent()
             return delegate.agent();
         }
 
-        @Override
-        public Id nodeId()
-        {
-            return delegate.nodeId();
-        }
-
         @Override
         public long minEpoch()
         {

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

@@ -49,6 +49,7 @@
 
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType;
+import org.apache.cassandra.service.accord.AccordService;
 import org.apache.cassandra.service.accord.serializers.ApplySerializers.ApplySerializer;
 import org.apache.cassandra.service.accord.serializers.IVersionedSerializer;
 import org.apache.cassandra.service.accord.txn.AccordUpdate;
@@ -71,7 +72,7 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL
         @Override
         public Apply create(Kind kind, Id to, Topologies participates, TxnId txnId, Ballot ballot, Route<?> route, Txn txn, Timestamp executeAt, Deps deps, Writes writes, Result result, FullRoute<?> fullRoute, ExecuteFlags flags)
         {
-            checkArgument(kind != Kind.Maximal, "Shouldn't need to send a maximal commit with interop support");
+            checkArgument(kind != Kind.Maximal || to.equals(AccordService.nodeId()), "Shouldn't need to send a maximal commit with interop support");
             ConsistencyLevel commitCL = txn.update() instanceof AccordUpdate ? ((AccordUpdate) txn.update()).cassandraCommitCL() : null;
             // Any asynchronous apply option should use the regular Apply that doesn't wait for writes to complete
             if (commitCL == null || commitCL == ConsistencyLevel.ANY)

src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java

@@ -228,7 +228,7 @@ private void repairViaAccordTransaction(DecoratedKey dk, Map<Replica, Mutation>
         Keys keys = Keys.of(partitionKey);
         // This is going create a new BlockingReadRepair inside an Accord transaction which will go down
         // the !isEventuallyConsistent path and apply the repairs through Accord command stores using AccordInteropExecution
-        UnrecoverableRepairUpdate<E, P> repairUpdate = new UnrecoverableRepairUpdate(AccordService.instance().nodeId(), this, keys, dk, accordMutations, writePlan);
+        UnrecoverableRepairUpdate<E, P> repairUpdate = new UnrecoverableRepairUpdate(AccordService.nodeId(), this, keys, dk, accordMutations, writePlan);
 
         /*
          * The motivation for using a read to apply read repair is that we want to apply the writes in the execute phase

src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java

@@ -887,7 +887,7 @@ public ClusterSimulation(RandomSource random, long seed, int uniqueNum,
                                    .set("commitlog_compression", new ParameterizedClass(LZ4Compressor.class.getName(), emptyMap()))
                                    .set("commitlog_sync", "batch")
                                    .set("accord.journal.flush_mode", "BATCH")
-                                   .set("accord.accurate_micros", "false")
+                                   .set("accord.precise_micros", "false")
                                    .set("accord.command_store_shard_count", "4");
 
                              if (memtableType.equals("TrieMemtable"))