diff --git a/modules/accord b/modules/accord index 01aff40402..e8c70e097f 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 01aff40402a55bada08f6e5753ee3a4010f407da +Subproject commit e8c70e097f305a5097084079f136d8be7c085b29 diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index 2e114186f1..b593a56299 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -1263,7 +1263,6 @@ public class AccordDebugKeyspace extends VirtualKeyspace @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()) diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 4df9dd4ce2..450a5c8451 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -309,6 +309,7 @@ public class AccordService implements IAccordService, Shutdownable // 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 class AccordService implements IAccordService, Shutdownable } else { + nodeId = new Node.Id(tcmId.id()); AccordService as = new AccordService(tcmIdToAccord(tcmId)); unsafeInstance = replyInstance = as; as.localStartup(); @@ -438,6 +440,11 @@ public class AccordService implements IAccordService, Shutdownable return i; } + public static Node.Id nodeId() + { + return nodeId; + } + public static boolean started() { if (!DatabaseDescriptor.getAccordTransactionsEnabled()) @@ -513,7 +520,7 @@ public class AccordService implements IAccordService, Shutdownable 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 class AccordService implements IAccordService, Shutdownable return scheduler; } - public Id nodeId() - { - return node.id(); - } - @Override public long minEpoch() { diff --git a/src/java/org/apache/cassandra/service/accord/IAccordService.java b/src/java/org/apache/cassandra/service/accord/IAccordService.java index 954f9ae66b..77296264e0 100644 --- a/src/java/org/apache/cassandra/service/accord/IAccordService.java +++ b/src/java/org/apache/cassandra/service/accord/IAccordService.java @@ -181,8 +181,6 @@ public interface IAccordService Agent agent(); - Id nodeId(); - long minEpoch(); void awaitDone(TableId id, long epoch); @@ -338,12 +336,6 @@ public interface IAccordService return null; } - @Override - public Id nodeId() - { - throw new UnsupportedOperationException(); - } - @Override public long minEpoch() { @@ -548,12 +540,6 @@ public interface IAccordService return delegate.agent(); } - @Override - public Id nodeId() - { - return delegate.nodeId(); - } - @Override public long minEpoch() { diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java index b4fde32365..5f09d68b51 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropApply.java @@ -49,6 +49,7 @@ import accord.utils.UnhandledEnum; 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) diff --git a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java index 3535f6ed50..a2e8c900d9 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java @@ -228,7 +228,7 @@ public class BlockingReadRepair, P extends ReplicaPlan.Fo 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 repairUpdate = new UnrecoverableRepairUpdate(AccordService.instance().nodeId(), this, keys, dk, accordMutations, writePlan); + UnrecoverableRepairUpdate 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 diff --git a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java index 749e0c0434..de90ddc855 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java +++ b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java @@ -887,7 +887,7 @@ public class ClusterSimulation implements AutoCloseable .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"))