From 74f4a5e42ab9d28d4c0b59ffea45e69850a9f692 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Wed, 22 Jul 2026 15:11:24 -0700 Subject: [PATCH] update --- .../service/accord/AccordService.java | 6 +- .../service/accord/journal/AccordJournal.java | 8 +- .../service/accord/journal/ReplayMarkers.java | 29 +++- .../AccordUniqueTimeStampsOnCrashTest.java | 124 ++++++++++++++++++ .../serializers/StopMarkerSerializerTest.java | 62 +++++++++ 5 files changed, 222 insertions(+), 7 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/accord/AccordUniqueTimeStampsOnCrashTest.java create mode 100644 test/unit/org/apache/cassandra/service/accord/serializers/StopMarkerSerializerTest.java diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 111e207474..5b8efdb487 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -554,6 +554,8 @@ public class AccordService implements IAccordService, Shutdownable rebootstrap = true; } } + + node.uniqueNow(ReplayMarkers.readLastUniqueTimeStamp()); } logger.info("Starting background compaction of system_accord"); @@ -684,7 +686,7 @@ public class AccordService implements IAccordService, Shutdownable // we set ourselves to STARTED before starting progress logs as this is the condition we use to decide if we // start the progress log on command store initialisation (so creates a synchronisation point) - journal.writeStartMarker(); + journal.writeStartMarker(node.uniqueNow()); state = State.STARTED; instance = requestInstance = this; node.commandStores().forAllUnsafe(cs -> cs.unsafeProgressLog().start()); @@ -1227,7 +1229,7 @@ public class AccordService implements IAccordService, Shutdownable AccordCommandStores commandStores = (AccordCommandStores)node.commandStores(); Set tableIds = commandStores.shutdownStores(); commandStores.waitForQuiescence(); - journal.writeSafeStopMarker(); + journal.writeSafeStopMarker(node.uniqueNow()); scheduler.shutdownNow(); toFuture(flushCaches()).map(ignore -> { return AccordColumnFamilyStores.commandsForKey.forceFlush(DRAIN); diff --git a/src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java index ee728a8778..b739a351f1 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java @@ -621,15 +621,15 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier return rangeSearch.rangeSearcher(); } - public void writeStartMarker() + public void writeStartMarker(long lastUniqueTimeStamp) { - writeMarker(startMarker(), segments.peekSegmentId()); + writeMarker(startMarker(), segments.peekSegmentId(), lastUniqueTimeStamp); } - public void writeSafeStopMarker() + public void writeSafeStopMarker(long lastUniqueTimeStamp) { segments.fsync(); - writeMarker(safeStopMarker(), segments.peekSegmentId()); + writeMarker(safeStopMarker(), segments.peekSegmentId(), lastUniqueTimeStamp); } private static Runnable merge(Runnable first, Runnable second) diff --git a/src/java/org/apache/cassandra/service/accord/journal/ReplayMarkers.java b/src/java/org/apache/cassandra/service/accord/journal/ReplayMarkers.java index 14845c31ae..8b0156537d 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/ReplayMarkers.java +++ b/src/java/org/apache/cassandra/service/accord/journal/ReplayMarkers.java @@ -41,11 +41,14 @@ public class ReplayMarkers } // TODO (required): add checksummed version and default to this (but support unchecksummed for manual editing) - static void writeMarker(File file, long timestamp) + public static void writeMarker(File file, long timestamp, long lastUniqueTimeStamp) { try (FileOutputStreamPlus out = new FileOutputStreamPlus(file)) { + int endingOffset = Long.toString(timestamp).length(); + out.writeInt(endingOffset); out.writeBytes(Long.toString(timestamp)); + out.writeBytes(Long.toString(lastUniqueTimeStamp)); } catch (IOException e) { @@ -72,6 +75,30 @@ public class ReplayMarkers try (FileInputStreamPlus in = new FileInputStreamPlus(file)) { StringBuilder sb = new StringBuilder(8); + int endingOffset = in.readInt(); + for (int i = 0; i < endingOffset; i++) + sb.append((char) in.read()); + return Long.parseLong(sb.toString()); + } + catch (IOException e) + { + throw new UncheckedIOException(e); + } + } + + public static long readLastUniqueTimeStamp() + { + File file = safeStopMarker(); + if (!file.exists()) + return -1L; + + try (FileInputStreamPlus in = new FileInputStreamPlus(file)) + { + int endingOffset = in.readInt(); + StringBuilder sb = new StringBuilder(8); + for (int i = 0; i < endingOffset; i++) + in.read(); + for (int b = in.read(); b >= 0 ; b = in.read()) sb.append((char)b); return Long.parseLong(sb.toString()); diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordUniqueTimeStampsOnCrashTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordUniqueTimeStampsOnCrashTest.java new file mode 100644 index 0000000000..59beaabc53 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordUniqueTimeStampsOnCrashTest.java @@ -0,0 +1,124 @@ +/* + * 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 org.apache.cassandra.distributed.test.accord; + +import accord.local.Node; +import org.apache.cassandra.distributed.test.TestBaseImpl; + +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; +import net.bytebuddy.implementation.bind.annotation.SuperCall; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.service.accord.AccordService; +import org.apache.cassandra.service.accord.api.AccordTimeService; +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.distributed.shared.NetworkTopology; +import org.apache.cassandra.utils.Shared; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + +public class AccordUniqueTimeStampsOnCrashTest extends TestBaseImpl +{ + @Test + public void uniqueTimeStampsOnCrashTest() throws Throwable + { + try (Cluster cluster = Cluster.build().withNodes(3) + .withoutVNodes() + .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(3)) + .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(3, "dc0", "rack0")) + .withInstanceInitializer(BBHelper::install) + .withConfig(config -> config + .with(NETWORK, GOSSIP)) + .start()) + { + cluster.schemaChange("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor':3}"); + cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key (k, c)) WITH transactional_mode='full'"); + cluster.coordinator(1).execute(wrapInTxn("INSERT INTO ks.tbl (k, c, v) VALUES (?, ?, ?)"), ConsistencyLevel.SERIAL, 1, 1, 2); + + cluster.get(1).shutdown(false).get(); + State.beforeRestart.set(false); + + cluster.get(1).startup(); + + cluster.get(1).runOnInstance( () -> { + AccordService.instance().node().uniqueNow(0); + }); + } + } + + @Shared + public static class State + { + public static AtomicBoolean beforeRestart = new AtomicBoolean(true); + public static AtomicLong timestamp = new AtomicLong(0); + } + + public static class BBHelper + { + + static void install(ClassLoader cl, int nodeNumber) + { + if (nodeNumber == 1) + { + new ByteBuddy().rebase(AccordTimeService.class) + .method(named("now").and(takesArguments(0))) + .intercept(MethodDelegation.to(BBHelper.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + + new ByteBuddy().rebase(Node.class) + .method(named("uniqueNow").and(takesArguments(1))) + .intercept(MethodDelegation.to(BBHelper.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + } + + public static long now(@SuperCall Callable r) throws Exception + { + if (State.beforeRestart.get()) + return r.call(); + + // Simulate clock skew on restart to be 100 seconds backwards + return r.call() - 100000000L; + } + + @SuppressWarnings("unused") + public static long uniqueNow(long greaterThan, @SuperCall Callable r) throws Exception + { + long newTimestamp = r.call(); + assert State.timestamp.get() < newTimestamp; + State.timestamp.set(newTimestamp); + + return r.call(); + } + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/StopMarkerSerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/StopMarkerSerializerTest.java new file mode 100644 index 0000000000..2f41ba0a0f --- /dev/null +++ b/test/unit/org/apache/cassandra/service/accord/serializers/StopMarkerSerializerTest.java @@ -0,0 +1,62 @@ +/* + * 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 org.apache.cassandra.service.accord.serializers; + +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import accord.utils.RandomSource; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.ServerTestUtils; +import org.apache.cassandra.config.CassandraRelevantProperties; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.io.util.File; +import org.apache.cassandra.service.accord.api.AccordTimeService; +import org.apache.cassandra.service.accord.journal.ReplayMarkers; +import org.apache.cassandra.utils.StorageCompatibilityMode; + +import static accord.utils.Property.qt; +import static org.apache.cassandra.service.accord.journal.ReplayMarkers.safeStopMarker; +import static org.apache.cassandra.service.accord.journal.ReplayMarkers.writeMarker; + +public class StopMarkerSerializerTest +{ + @BeforeClass + public static void beforeClass() throws Throwable + { + CassandraRelevantProperties.TEST_STORAGE_COMPATIBILITY_MODE.setEnum(StorageCompatibilityMode.NONE); + SchemaLoader.prepareServer(); + } + + @Test + public void stopMarkerSerializerTest() + { + if (new File(DatabaseDescriptor.getAccordJournalDirectory()).exists()) + ServerTestUtils.cleanupDirectory(DatabaseDescriptor.getAccordJournalDirectory()); + + qt().forAll(RandomSource::nextLong).check(timestamp -> { + long lastUniqueTimeStamp = AccordTimeService.nowMicros(); + writeMarker(safeStopMarker(), timestamp, lastUniqueTimeStamp); + Assert.assertEquals(timestamp.longValue(), ReplayMarkers.readStopMarker()); + Assert.assertEquals(lastUniqueTimeStamp, ReplayMarkers.readLastUniqueTimeStamp()); + }); + } +} \ No newline at end of file