natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

update

This commit is contained in:
Alan Wang 2026-07-22 15:11:24 -07:00
parent e805b8aa32
commit 74f4a5e42a
5 changed files with 222 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -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<TableId> tableIds = commandStores.shutdownStores();
commandStores.waitForQuiescence();
journal.writeSafeStopMarker();
journal.writeSafeStopMarker(node.uniqueNow());
scheduler.shutdownNow();
toFuture(flushCaches()).map(ignore -> {
return AccordColumnFamilyStores.commandsForKey.forceFlush(DRAIN);

8
src/java/org/apache/cassandra/service/accord/journal/AccordJournal.java
View File

@ -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)

29
src/java/org/apache/cassandra/service/accord/journal/ReplayMarkers.java
View File

@ -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());

124
test/distributed/org/apache/cassandra/distributed/test/accord/AccordUniqueTimeStampsOnCrashTest.java Normal file
View File

@ -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<Long> 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<Long> r) throws Exception
{
long newTimestamp = r.call();
assert State.timestamp.get() < newTimestamp;
State.timestamp.set(newTimestamp);
return r.call();
}
}
}

62
test/unit/org/apache/cassandra/service/accord/serializers/StopMarkerSerializerTest.java Normal file
View File

@ -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());
});
}
}