This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-45-mutation-tracking by
this push:
new 38e01d0972 CEP-45: Last coordinator log id can decrement on startup
38e01d0972 is described below
commit 38e01d09729aa01ff262c6b9c1cd03df210d2b70
Author: Blake Eggleston <[email protected]>
AuthorDate: Mon Oct 19 15:24:06 2026 -0700
CEP-45: Last coordinator log id can decrement on startup
- prevents broadcast log offsets messages from being processed prior to MTS
startup
- removes eager shard creation
- fixes a few other concurrency / durability gaps
Patch by Blake Eggleston; Reviewed by Caleb Rackliffe for CASSANDRA-21790
---
.../db/commitlog/AbstractCommitLogService.java | 4 +-
.../apache/cassandra/db/commitlog/CommitLog.java | 8 +
.../replication/MutationTrackingService.java | 234 +++++++++++++++------
.../replication/MutationTrackingServiceTest.java | 171 ++++++++++++++-
.../apache/cassandra/db/CleanupTransientTest.java | 1 +
.../db/CoordinatorLogOffsetsLifecycleTest.java | 3 +
.../replication/CoordinatorLogOffsetsTest.java | 2 +-
.../replication/MutationTrackingServiceTest.java | 51 +++++
.../replication/PendingLocalTransferTest.java | 2 +
.../replication/TrackedCounterWriteTest.java | 2 +
10 files changed, 410 insertions(+), 68 deletions(-)
diff --git
a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
index 1831ec866e..7359cc95a4 100644
--- a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
+++ b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
@@ -319,12 +319,12 @@ public abstract class AbstractCommitLogService
do
{
WaitQueue.Signal signal = context != null ?
syncComplete.register(context, Context::stop) : syncComplete.register();
- if (lastSyncedAt < syncTime)
+ if (lastSyncedAt - syncTime < 0)
signal.awaitUninterruptibly();
else
signal.cancel();
}
- while (lastSyncedAt < syncTime);
+ while (lastSyncedAt - syncTime < 0);
}
public void awaitTermination() throws InterruptedException
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
index b47bf7804b..7c8ad49c70 100644
--- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
+++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
@@ -292,6 +292,14 @@ public class CommitLog implements CommitLogMBean
executor.requestExtraSync();
}
+ /**
+ * Request sync and wait until the current state is synced.
+ */
+ public void syncBlocking()
+ {
+ executor.syncBlocking();
+ }
+
/**
* Add a Mutation to the commit log. If CDC is enabled, this can fail.
*
diff --git
a/src/java/org/apache/cassandra/replication/MutationTrackingService.java
b/src/java/org/apache/cassandra/replication/MutationTrackingService.java
index 7cbc31fc3f..b4ec550b14 100644
--- a/src/java/org/apache/cassandra/replication/MutationTrackingService.java
+++ b/src/java/org/apache/cassandra/replication/MutationTrackingService.java
@@ -60,6 +60,7 @@ import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.Range;
@@ -81,6 +82,7 @@ import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.reads.tracked.TrackedLocalReads;
import org.apache.cassandra.streaming.StreamOperation;
import org.apache.cassandra.tcm.ClusterMetadata;
@@ -224,6 +226,9 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
private final Map<String, Set<MutationTrackingSyncCoordinator>>
syncCoordinatorsByKeyspace = new ConcurrentHashMap<>();
+ private final Object hostLogIdLock = new Object();
+ private volatile boolean hostLogIdLoaded = false;
+
private volatile boolean started = false;
private MutationTrackingService()
@@ -243,11 +248,11 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
if (started)
return;
- prevHostLogId = loadHostLogIdFromSystemTable();
+ ClusterMetadata metadata = register.apply(tcmListener);
- logger.info("Starting mutation tracking service. Previous host log id:
{}", prevHostLogId);
+ initHostLogId(metadata);
- ClusterMetadata metadata = register.apply(tcmListener);
+ logger.info("Starting mutation tracking service. Previous host log id:
{}", prevHostLogId);
if (metadata.myNodeId() != null)
for (KeyspaceShards ks :
KeyspaceShards.loadFromSystemTables(metadata, this::nextLogId, this::onNewLog))
@@ -371,7 +376,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
shardLock.readLock().lock();
try
{
- MutationId id = getOrCreateShards(keyspace).nextMutationId(token);
+ MutationId id = getShards(keyspace).nextMutationId(token);
logger.trace("Created new mutation id {}", id);
return id;
}
@@ -387,7 +392,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
shardLock.readLock().lock();
try
{
- KeyspaceShards shards = getOrCreateShards(keyspace);
+ KeyspaceShards shards = getShards(keyspace);
Shard shard = null;
for (Range<Token> range : ranges)
{
@@ -477,7 +482,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
shardLock.readLock().lock();
try
{
- KeyspaceShards shards = maybeGetOrCreateShards(keyspace);
+ KeyspaceShards shards = keyspaceShards.get(keyspace);
if (shards != null)
shards.updateReplicatedOffsets(range, offsets, durable,
onHost);
else
@@ -508,7 +513,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
try
{
reconciledSnapshot.forEach((keyspace, keyspaceOffsets) -> {
- KeyspaceShards ksShards = maybeGetOrCreateShards(keyspace);
+ KeyspaceShards ksShards = keyspaceShards.get(keyspace);
if (ksShards != null)
ksShards.recordFullyReconciledOffsets(keyspaceOffsets);
else
@@ -527,7 +532,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
try
{
Preconditions.checkArgument(!mutation.id().isNone());
- boolean started =
getOrCreateShards(mutation.getKeyspaceName()).startWriting(mutation);
+ boolean started =
getShards(mutation.getKeyspaceName()).startWriting(mutation);
// we've already received this mutation, so we reject the write.
There may still be reconciliation
// listening for it though, so notify them that it's received
if (!started)
@@ -582,7 +587,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
try
{
Preconditions.checkArgument(!mutation.id().isNone());
-
getOrCreateShards(mutation.getKeyspaceName()).finishWriting(mutation);
+ getShards(mutation.getKeyspaceName()).finishWriting(mutation);
incomingMutations.invokeListeners(mutation.id());
}
finally
@@ -713,7 +718,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
shardLock.readLock().lock();
try
{
- MutationSummary summary =
getOrCreateShards(tableId).createSummaryForKey(key, tableId, includePending);
+ MutationSummary summary =
getShards(tableId).createSummaryForKey(key, tableId, includePending);
MutationTrackingMetrics.instance().readSummarySize.update(summary.size());
return summary;
}
@@ -728,7 +733,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
shardLock.readLock().lock();
try
{
- MutationSummary summary =
getOrCreateShards(tableId).createSummaryForRange(range, tableId,
includePending);
+ MutationSummary summary =
getShards(tableId).createSummaryForRange(range, tableId, includePending);
MutationTrackingMetrics.instance().readSummarySize.update(summary.size());
return summary;
}
@@ -913,33 +918,22 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
return Preconditions.checkNotNull(log2ShardMap.get(logId));
}
- private KeyspaceShards getOrCreateShards(TableId tableId)
+ private KeyspaceShards getShards(TableId tableId)
{
- //noinspection DataFlowIssue
- return
getOrCreateShards(Schema.instance.getTableMetadata(tableId).keyspace);
+ TableMetadata metadata = Schema.instance.getTableMetadata(tableId);
+ if (metadata == null)
+ throw new IllegalStateException(String.format("Unknown table %s",
tableId));
+ return getShards(metadata.keyspace);
}
- private KeyspaceShards getOrCreateShards(String keyspace)
+ private KeyspaceShards getShards(String keyspace)
{
- KeyspaceShards shards = maybeGetOrCreateShards(keyspace);
- Preconditions.checkArgument(shards != null, "Unknown keyspace %s",
keyspace);
+ KeyspaceShards shards = keyspaceShards.get(keyspace);
+ if (shards == null)
+ throw new IllegalStateException(String.format("No mutation
tracking shards found for keyspace %s", keyspace));
return shards;
}
- private KeyspaceShards maybeGetOrCreateShards(String keyspace)
- {
- KeyspaceShards ks = keyspaceShards.get(keyspace);
- if (ks != null)
- return ks;
-
- ClusterMetadata csm = ClusterMetadata.current();
- KeyspaceMetadata ksm =
csm.schema.maybeGetKeyspaceMetadata(keyspace).orElse(null);
- if (ksm == null)
- return null;
-
- return keyspaceShards.computeIfAbsent(keyspace, ignore ->
KeyspaceShards.make(ksm, csm, this::nextLogId, this::onNewLog));
- }
-
private long nextLogId()
{
NodeId nodeId = ClusterMetadata.current().myNodeId();
@@ -953,11 +947,15 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
*/
private int nextHostLogId()
{
- int nextHostLogId = ++prevHostLogId;
- persistHostLogIdToSystemTable(nextHostLogId);
- return nextHostLogId;
+ synchronized (hostLogIdLock)
+ {
+ Preconditions.checkState(hostLogIdLoaded, "Host log id counter has
not been loaded from system tables");
+ int nextHostLogId = ++prevHostLogId;
+ persistHostLogIdToSystemTable(nextHostLogId);
+ return nextHostLogId;
+ }
}
- private int prevHostLogId;
+ private volatile int prevHostLogId;
public boolean isDurablyReconciled(ShortMutationId id)
{
@@ -1199,39 +1197,47 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
{
Preconditions.checkArgument(keyspace.getMetadata().replicationStrategy.replicationType.isTracked(),
"Keyspace " + keyspace.getName() + " is not tracked");
- KeyspaceShards shards = keyspaceShards.get(keyspace.getName());
- Map<Shard, List<SyncTask>> tasksByShard = new HashMap<>();
-
- // Shard ranges do not wrap, so unwrap the task ranges before we start
comparing them.
- for (SyncTask task : unwrapped(tasks))
+ shardLock.readLock().lock();
+ try
{
- Set<Shard> intersectingShards = new HashSet<>();
- shards.forEachIntersectingShard(task.rangesToSync,
intersectingShards::add);
- for (Shard shard : intersectingShards)
+ KeyspaceShards shards = getShards(keyspace.getName());
+ Map<Shard, List<SyncTask>> tasksByShard = new HashMap<>();
+
+ // Shard ranges do not wrap, so unwrap the task ranges before we
start comparing them.
+ for (SyncTask task : unwrapped(tasks))
{
- // Ensure that we don't expand outside the ranges of the
original sync tasks.
- Set<Range<Token>> intersectingSyncRanges = new HashSet<>();
- for (Range<Token> syncRange : task.rangesToSync)
-
intersectingSyncRanges.addAll(syncRange.intersectionWith(shard.range));
+ Set<Shard> intersectingShards = new HashSet<>();
+ shards.forEachIntersectingShard(task.rangesToSync,
intersectingShards::add);
+ for (Shard shard : intersectingShards)
+ {
+ // Ensure that we don't expand outside the ranges of the
original sync tasks.
+ Set<Range<Token>> intersectingSyncRanges = new HashSet<>();
+ for (Range<Token> syncRange : task.rangesToSync)
+
intersectingSyncRanges.addAll(syncRange.intersectionWith(shard.range));
- if (!intersectingSyncRanges.isEmpty())
- tasksByShard.computeIfAbsent(shard, key -> new
ArrayList<>()).add(task.withRanges(intersectingSyncRanges));
+ if (!intersectingSyncRanges.isEmpty())
+ tasksByShard.computeIfAbsent(shard, key -> new
ArrayList<>()).add(task.withRanges(intersectingSyncRanges));
+ }
}
- }
- SyncTasks into = new SyncTasks();
+ SyncTasks into = new SyncTasks();
- for (Map.Entry<Shard, List<SyncTask>> entry : tasksByShard.entrySet())
- {
- Shard shard = entry.getKey();
- Collection<SyncTask> syncTasks = entry.getValue();
+ for (Map.Entry<Shard, List<SyncTask>> entry :
tasksByShard.entrySet())
+ {
+ Shard shard = entry.getKey();
+ Collection<SyncTask> syncTasks = entry.getValue();
- // Assign a new transfer ID to each sync task and add to the tasks
container
- for (SyncTask task : syncTasks)
- into.add(shard, task.withTransferId(shard.nextId()));
- }
+ // Assign a new transfer ID to each sync task and add to the
tasks container
+ for (SyncTask task : syncTasks)
+ into.add(shard, task.withTransferId(shard.nextId()));
+ }
- return into;
+ return into;
+ }
+ finally
+ {
+ shardLock.readLock().unlock();
+ }
}
private static List<SyncTask> unwrapped(Collection<SyncTask> tasks)
@@ -1446,7 +1452,10 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
MutationId nextMutationId(Token token)
{
- return lookUp(token).nextId();
+ Shard shard = lookUp(token);
+ if (shard == null)
+ throw new IllegalStateException(String.format("No mutation
tracking shard covers token %s in keyspace %s", token, keyspace));
+ return shard.nextId();
}
void updateReplicatedOffsets(Range<Token> range, List<? extends
Offsets> offsets, boolean durable, InetAddressAndPort onHost)
@@ -1459,7 +1468,10 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
boolean startWriting(Mutation mutation)
{
- return lookUp(mutation).startWriting(mutation);
+ Shard shard = lookUp(mutation);
+ if (shard == null)
+ throw new IllegalStateException(String.format("No mutation
tracking shard covers mutation %s in keyspace %s", mutation.key(), keyspace));
+ return shard.startWriting(mutation);
}
/**
@@ -1481,13 +1493,19 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
void finishWriting(Mutation mutation)
{
- lookUp(mutation).finishWriting(mutation);
+ Shard shard = lookUp(mutation);
+ if (shard == null)
+ throw new IllegalStateException(String.format("No mutation
tracking shard covers mutation %s in keyspace %s", mutation.key(), keyspace));
+ shard.finishWriting(mutation);
}
MutationSummary createSummaryForKey(DecoratedKey key, TableId tableId,
boolean includePending)
{
MutationSummary.Builder builder = new
MutationSummary.Builder(tableId);
- lookUp(key.getToken()).addSummaryForKey(key.getToken(),
includePending, builder);
+ Shard shard = lookUp(key.getToken());
+ if (shard == null)
+ throw new IllegalStateException(String.format("No mutation
tracking shard covers key %s in keyspace %s", key.getToken(), keyspace));
+ shard.addSummaryForKey(key.getToken(), includePending, builder);
return builder.build();
}
@@ -1605,6 +1623,7 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
static void persistHostLogIdToSystemTable(int hostLogId)
{
executeInternal(INSERT_QUERY, HOST_LOG_ID_KEY, hostLogId);
+ CommitLog.instance.syncBlocking();
}
private static final String SELECT_QUERY =
@@ -1618,6 +1637,46 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
return rows.one().getInt("host_log_id");
}
+ private static final String SELECT_COORDINATOR_LOGS_QUERY =
+ format("SELECT host_id, host_log_id FROM %s.%s",
+ SchemaConstants.SYSTEM_KEYSPACE_NAME,
SystemKeyspace.COORDINATOR_LOGS);
+
+ static int loadMaxCoordinatorLogIdFromSystemTable(int localHostId)
+ {
+ UntypedResultSet rows = executeInternal(SELECT_COORDINATOR_LOGS_QUERY);
+ int max = 0;
+ for (UntypedResultSet.Row row : rows)
+ {
+ if (row.has("host_id") && row.getInt("host_id") == localHostId)
+ {
+ if (row.has("host_log_id"))
+ max = Math.max(max, row.getInt("host_log_id"));
+ }
+ }
+ return max;
+ }
+
+ private void initHostLogId(ClusterMetadata metadata)
+ {
+ NodeId myNodeId = metadata.myNodeId();
+ synchronized (hostLogIdLock)
+ {
+ int storedHostLogId = loadHostLogIdFromSystemTable();
+ int maxCoordinatorLogId = myNodeId != null ?
loadMaxCoordinatorLogIdFromSystemTable(myNodeId.id()) : 0;
+ if (maxCoordinatorLogId > storedHostLogId)
+ {
+ logger.warn("Stored mutation tracking host log id {} in {}.{}
has regressed; highest log id found in {}.{} for local node {} is {}",
+ storedHostLogId,
SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.HOST_LOG_ID,
+ SchemaConstants.SYSTEM_KEYSPACE_NAME,
SystemKeyspace.COORDINATOR_LOGS,
+ myNodeId, maxCoordinatorLogId);
+ }
+ prevHostLogId = Math.max(storedHostLogId, maxCoordinatorLogId);
+ if (prevHostLogId > storedHostLogId)
+ persistHostLogIdToSystemTable(prevHostLogId);
+ hostLogIdLoaded = true;
+ }
+ }
+
private static class BackgroundReconciler
{
void start()
@@ -1927,10 +1986,57 @@ public class MutationTrackingService implements
MutationTrackingServiceMBean
public static class TestAccess
{
public static MutationTrackingService create()
+ {
+ MutationTrackingService service = new MutationTrackingService();
+ service.hostLogIdLoaded = true;
+ return service;
+ }
+
+ public static MutationTrackingService createUnloaded()
{
return new MutationTrackingService();
}
+ public static int nextHostLogId(MutationTrackingService service)
+ {
+ return service.nextHostLogId();
+ }
+
+ public static void initHostLogId(MutationTrackingService service,
ClusterMetadata metadata)
+ {
+ service.initHostLogId(metadata);
+ }
+
+ public static int getPrevHostLogId(MutationTrackingService service)
+ {
+ return service.prevHostLogId;
+ }
+
+ public static List<Long> getPrimaryHostLogIds(MutationTrackingService
service, String keyspace)
+ {
+ KeyspaceShards shards = service.keyspaceShards.get(keyspace);
+ List<Long> ids = new ArrayList<>();
+ if (shards != null)
+ shards.forEachShard(s -> ids.add((long)
s.currentLocalLog().logId.hostLogId));
+ return ids;
+ }
+
+ public static void broadcastOffsetsTo(MutationTrackingService service,
String keyspace, InetAddressAndPort target)
+ {
+ KeyspaceShards shards = service.keyspaceShards.get(keyspace);
+ if (shards != null)
+ {
+ shards.forEachShard(shard -> {
+ BroadcastLogOffsets transientOffsets =
shard.collectReplicatedOffsets(false);
+ if (!transientOffsets.isEmpty())
+
MessagingService.instance().send(Message.out(Verb.MT_BROADCAST_LOG_OFFSETS,
transientOffsets), target);
+ BroadcastLogOffsets durableOffsets =
shard.collectReplicatedOffsets(true);
+ if (!durableOffsets.isEmpty())
+
MessagingService.instance().send(Message.out(Verb.MT_BROADCAST_LOG_OFFSETS,
durableOffsets), target);
+ });
+ }
+ }
+
public static KeyspaceShards getKeyspaceShards(MutationTrackingService
service, String keyspace)
{
return service.keyspaceShards.get(keyspace);
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingServiceTest.java
b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingServiceTest.java
index 055b1e4ead..6b77b2a0f7 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingServiceTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingServiceTest.java
@@ -18,12 +18,30 @@
package org.apache.cassandra.distributed.test.replication;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.replication.MutationTrackingService;
+import org.apache.cassandra.utils.Shared;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
public class MutationTrackingServiceTest extends TestBaseImpl
@@ -31,6 +49,56 @@ public class MutationTrackingServiceTest extends TestBaseImpl
private static final String KS_NAME = "ks";
private static final String TBL_NAME = "tbl";
+ @Shared
+ public static class BBState
+ {
+ public static volatile boolean enabled = false;
+ public static volatile CountDownLatch enteredLatch = new
CountDownLatch(1);
+ public static volatile CountDownLatch pauseLatch = new
CountDownLatch(1);
+
+ public static void reset()
+ {
+ enabled = false;
+ enteredLatch = new CountDownLatch(1);
+ pauseLatch = new CountDownLatch(1);
+ }
+ }
+
+ public static class BBInstaller
+ {
+ public static void install(ClassLoader cl, Integer num)
+ {
+ if (num == 1)
+ {
+ new ByteBuddy().rebase(MutationTrackingService.class)
+ .method(named("start").and(takesArguments(0)))
+
.intercept(MethodDelegation.to(BBInterceptor.class))
+ .make()
+ .load(cl,
ClassLoadingStrategy.Default.INJECTION);
+ }
+ }
+ }
+
+ public static class BBInterceptor
+ {
+ public static void start(@SuperCall Runnable zuper)
+ {
+ if (BBState.enabled)
+ {
+ BBState.enteredLatch.countDown();
+ try
+ {
+ BBState.pauseLatch.await(60, TimeUnit.SECONDS);
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+ zuper.run();
+ }
+ }
+
@Test
public void testIrrelevantTopologyChange() throws Throwable
{
@@ -52,8 +120,109 @@ public class MutationTrackingServiceTest extends
TestBaseImpl
Object newShards =
MutationTrackingService.TestAccess.getKeyspaceShards(service, KS_NAME);
return initialShards == newShards; // Same object reference
});
-
+
assertTrue("Keyspace shards should not change for irrelevant
topology changes", shardsUnchanged);
}
}
+
+ @Test
+ public void testHostLogIdNotRegressedByBroadcastDuringStartup() throws
Throwable
+ {
+ BBState.reset();
+ try (Cluster cluster = builder().withNodes(2)
+
.withInstanceInitializer(BBInstaller::install)
+ .start())
+ {
+ cluster.schemaChange("CREATE KEYSPACE " + KS_NAME + " WITH
replication = " +
+ "{'class': 'SimpleStrategy',
'replication_factor': 2} AND replication_type='tracked'");
+ cluster.schemaChange("CREATE TABLE " + KS_NAME + "." + TBL_NAME +
" (k int PRIMARY KEY, v int)");
+
+ // Perform writes to generate mutations from both coordinators so
shards have active offsets
+ cluster.coordinator(1).execute("INSERT INTO " + KS_NAME + "." +
TBL_NAME + " (k, v) VALUES (1, 1)",
+ ConsistencyLevel.ALL);
+ cluster.coordinator(2).execute("INSERT INTO " + KS_NAME + "." +
TBL_NAME + " (k, v) VALUES (2, 2)",
+ ConsistencyLevel.ALL);
+
+ // Allocate additional log IDs on node 1 so counter is well above
1 (mirroring a node that ran for some time)
+ cluster.get(1).runOnInstance(() -> {
+ var service = MutationTrackingService.instance();
+ for (int i = 0; i < 10; i++)
+ MutationTrackingService.TestAccess.nextHostLogId(service);
+ });
+
+ // Read host_log_id before restart on node 1
+ Object[][] res = cluster.get(1).executeInternal("SELECT
host_log_id FROM system.host_log_id WHERE key='local'");
+ assertNotNull(res);
+ assertTrue(res.length > 0);
+ int initialHostLogId = (Integer) res[0][0];
+ assertTrue("Expected positive initial host log id, got " +
initialHostLogId, initialHostLogId > 0);
+
+ // Ensure node 2 has witnessed the mutations and can broadcast
offsets
+ cluster.get(2).runOnInstance(() -> {
+
MutationTrackingService.instance().broadcastOffsetsForTesting();
+ });
+
+ // Shut down node 1
+ cluster.get(1).shutdown().get();
+
+ // Enable startup pause for node 1
+ BBState.enabled = true;
+ BBState.enteredLatch = new CountDownLatch(1);
+ BBState.pauseLatch = new CountDownLatch(1);
+
+ // Start node 1 asynchronously
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ Future<?> startupFuture = executor.submit(() -> {
+ try
+ {
+ cluster.get(1).startup();
+ }
+ catch (Throwable t)
+ {
+ throw new RuntimeException(t);
+ }
+ });
+
+ try
+ {
+ // Wait for node 1 to reach MutationTrackingService.start()
+ assertTrue("Timed out waiting for node 1 to reach
MutationTrackingService.start()",
+ BBState.enteredLatch.await(60, TimeUnit.SECONDS));
+
+ InetAddressAndPort node1Addr =
InetAddressAndPort.getByNameUnchecked("127.0.0.1:7012");
+ cluster.get(2).runOnInstance(() -> {
+
MutationTrackingService.TestAccess.broadcastOffsetsTo(MutationTrackingService.instance(),
KS_NAME, node1Addr);
+ });
+ Thread.sleep(500);
+ }
+ finally
+ {
+ // Unpause node 1 so startup can finish
+ BBState.pauseLatch.countDown();
+ }
+
+ startupFuture.get(60, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // Verify post-restart state on node 1:
+ // 1. system.host_log_id must be strictly greater than
initialHostLogId (since boot allocates a new primary log)
+ Object[][] postRes = cluster.get(1).executeInternal("SELECT
host_log_id FROM system.host_log_id WHERE key='local'");
+ assertNotNull(postRes);
+ assertTrue(postRes.length > 0);
+ int postHostLogId = (Integer) postRes[0][0];
+ assertTrue("Post restart host log id " + postHostLogId + " should
be greater than initial " + initialHostLogId,
+ postHostLogId > initialHostLogId);
+
+ // 2. New primary log id on each shard must be strictly greater
than initialHostLogId
+ List<Long> newPrimaryLogIds = cluster.get(1).callOnInstance(() -> {
+ return
MutationTrackingService.TestAccess.getPrimaryHostLogIds(MutationTrackingService.instance(),
KS_NAME);
+ });
+
+ for (Long logId : newPrimaryLogIds)
+ {
+ assertTrue("New primary log id " + logId + " must be greater
than initial host log id " + initialHostLogId,
+ logId > initialHostLogId);
+ }
+ }
+ }
}
diff --git a/test/unit/org/apache/cassandra/db/CleanupTransientTest.java
b/test/unit/org/apache/cassandra/db/CleanupTransientTest.java
index b560f6a300..a2c58bd8d6 100644
--- a/test/unit/org/apache/cassandra/db/CleanupTransientTest.java
+++ b/test/unit/org/apache/cassandra/db/CleanupTransientTest.java
@@ -89,6 +89,7 @@ public class CleanupTransientTest extends CassandraTestBase
Util.createInitialRing(endpointTokens, keyTokens, hosts, hostIds,
RING_SIZE);
MutationJournal.start();
+ MutationTrackingService.start();
SchemaLoader.createKeyspace(KEYSPACE1,
KeyspaceParams.simpleWitness("2/1"),
SchemaLoader.standardCFMD(KEYSPACE1,
CF_STANDARD1),
diff --git
a/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
b/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
index bc6f2b2046..31a5ea32fc 100644
--- a/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
+++ b/test/unit/org/apache/cassandra/db/CoordinatorLogOffsetsLifecycleTest.java
@@ -45,6 +45,7 @@ import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.replication.MutationId;
+import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.MemtableParams;
@@ -68,6 +69,8 @@ public class CoordinatorLogOffsetsLifecycleTest
public static void setupClass()
{
SchemaLoader.prepareServer();
+ MutationJournal.start();
+ MutationTrackingService.start();
}
@Parameter
diff --git
a/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
b/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
index ca6b63298f..3cc992f0b6 100644
--- a/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
+++ b/test/unit/org/apache/cassandra/replication/CoordinatorLogOffsetsTest.java
@@ -278,7 +278,7 @@ public class CoordinatorLogOffsetsTest
Keyspace.setInitialized();
ClusterMetadata.current().schema.initializeKeyspaceInstances(DistributedSchema.empty(),
false);
- MutationTrackingService.start(unused -> metadata);
+ MutationTrackingService.start(unused -> ClusterMetadata.current());
// Eventually, will also run perturbations before checking
isReconciled (like log truncation, durability, etc.)
// to ensure that we don't prune data required to check what's been
reconciled
diff --git
a/test/unit/org/apache/cassandra/replication/MutationTrackingServiceTest.java
b/test/unit/org/apache/cassandra/replication/MutationTrackingServiceTest.java
index ce9e61f40a..52f91b254f 100644
---
a/test/unit/org/apache/cassandra/replication/MutationTrackingServiceTest.java
+++
b/test/unit/org/apache/cassandra/replication/MutationTrackingServiceTest.java
@@ -35,8 +35,10 @@ import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.Range;
@@ -50,6 +52,7 @@ import org.apache.cassandra.repair.SyncTask;
import org.apache.cassandra.repair.SyncTasks;
import org.apache.cassandra.replication.MutationTrackingService.KeyspaceShards;
import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.ReplicationType;
import org.apache.cassandra.schema.SchemaTestUtil;
import org.apache.cassandra.schema.TableId;
@@ -443,6 +446,54 @@ public class MutationTrackingServiceTest
}
}
+ @Test
+ public void testNextHostLogIdFailsBeforeLoaded()
+ {
+ MutationTrackingService service =
MutationTrackingService.TestAccess.createUnloaded();
+ try
+ {
+ MutationTrackingService.TestAccess.nextHostLogId(service);
+ Assert.fail("Expected IllegalStateException when allocating log ID
before counter is loaded");
+ }
+ catch (IllegalStateException e)
+ {
+ assertTrue(e.getMessage().contains("Host log id counter has not
been loaded"));
+ }
+ }
+
+ @Test
+ public void testStartupGuardBumpsRegressedCounter()
+ {
+ QueryProcessor.executeInternal(String.format("INSERT INTO %s.%s
(keyspace_name, range_start, range_end, host_id, host_log_id) VALUES (?, ?, ?,
?, ?)",
+
SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.COORDINATOR_LOGS),
+ "test_ks", "a", "z", 1, 56);
+ QueryProcessor.executeInternal(String.format("INSERT INTO %s.%s
(keyspace_name, range_start, range_end, host_id, host_log_id) VALUES (?, ?, ?,
?, ?)",
+
SchemaConstants.SYSTEM_KEYSPACE_NAME, SystemKeyspace.COORDINATOR_LOGS),
+ "test_ks", "a", "z", 2, 100);
+
+ assertEquals(56,
MutationTrackingService.loadMaxCoordinatorLogIdFromSystemTable(1));
+ assertEquals(100,
MutationTrackingService.loadMaxCoordinatorLogIdFromSystemTable(2));
+
+ // Stored counter is 6 (regressed compared to 56)
+ MutationTrackingService.persistHostLogIdToSystemTable(6);
+ assertEquals(6,
MutationTrackingService.loadHostLogIdFromSystemTable());
+
+ MutationTrackingService service =
MutationTrackingService.TestAccess.createUnloaded();
+ ClusterMetadata metadata = ClusterMetadata.current();
+ MutationTrackingService.TestAccess.initHostLogId(service, metadata);
+
+ // Counter bumped to 56 and persisted
+ assertEquals(56,
MutationTrackingService.TestAccess.getPrevHostLogId(service));
+ assertEquals(56,
MutationTrackingService.loadHostLogIdFromSystemTable());
+
+ // When stored is higher, counter does not regress
+ MutationTrackingService.persistHostLogIdToSystemTable(70);
+ MutationTrackingService service2 =
MutationTrackingService.TestAccess.createUnloaded();
+ MutationTrackingService.TestAccess.initHostLogId(service2, metadata);
+ assertEquals(70,
MutationTrackingService.TestAccess.getPrevHostLogId(service2));
+ assertEquals(70,
MutationTrackingService.loadHostLogIdFromSystemTable());
+ }
+
private static Token tk(String key)
{
return new
ByteOrderedPartitioner.BytesToken(ByteBufferUtil.bytes(key));
diff --git
a/test/unit/org/apache/cassandra/replication/PendingLocalTransferTest.java
b/test/unit/org/apache/cassandra/replication/PendingLocalTransferTest.java
index 751af68497..604dedfc77 100644
--- a/test/unit/org/apache/cassandra/replication/PendingLocalTransferTest.java
+++ b/test/unit/org/apache/cassandra/replication/PendingLocalTransferTest.java
@@ -78,6 +78,8 @@ public class PendingLocalTransferTest
public static void setUpClass()
{
SchemaLoader.prepareServer();
+ MutationJournal.start();
+ MutationTrackingService.start();
TableMetadata metadata = CreateTableStatement.parse(format("CREATE
TABLE %s.%s (k int PRIMARY KEY, v int)", KS, TBL), KS).build();
SchemaLoader.createKeyspace(KS, KeyspaceParams.simple(1,
ReplicationType.tracked), metadata);
cfs = Schema.instance.getColumnFamilyStoreInstance(metadata.id);
diff --git
a/test/unit/org/apache/cassandra/replication/TrackedCounterWriteTest.java
b/test/unit/org/apache/cassandra/replication/TrackedCounterWriteTest.java
index db0f3df2aa..a2ec95af40 100644
--- a/test/unit/org/apache/cassandra/replication/TrackedCounterWriteTest.java
+++ b/test/unit/org/apache/cassandra/replication/TrackedCounterWriteTest.java
@@ -59,6 +59,8 @@ public class TrackedCounterWriteTest
public static void defineSchema() throws ConfigurationException
{
SchemaLoader.prepareServer();
+ MutationJournal.start();
+ MutationTrackingService.start();
SchemaLoader.createKeyspace(KEYSPACE_TRACKED,
KeyspaceParams.simple(3,
ReplicationType.tracked),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
