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

CEP-45: Background reconciliation - replica 

patch by Francisco Guerrero; reviewed by Blake Eggleston for CASSANDRA-20378
This commit is contained in:
Francisco Guerrero 2026-04-19 13:06:45 -07:00 committed by Francisco Guerrero
parent 16bb95fa15
commit c6814f3c69
25 changed files with 782 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -6059,6 +6059,11 @@ public class DatabaseDescriptor
}
}
public static MutationTrackingSpec getMutationTrackingConfig()
{
return conf.mutation_tracking;
}
public static boolean getMutationTrackingEnabled()
{
return conf.mutation_tracking.enabled;

8
src/java/org/apache/cassandra/config/MutationTrackingSpec.java
View File

@ -22,4 +22,12 @@ public class MutationTrackingSpec
{
public boolean enabled = false;
public String journal_directory;
/**
* Whether the background reconciliation process is enabled
*/
public volatile boolean background_reconciliation_enabled = true;
/**
* The interval in which the backgroun reconciliation process runs
*/
public volatile DurationSpec.LongMillisecondsBound background_reconciliation_interval = new DurationSpec.LongMillisecondsBound("1s");
}

1
src/java/org/apache/cassandra/replication/ActiveLogReconciler.java
View File

@ -173,6 +173,7 @@ public final class ActiveLogReconciler implements Shutdownable
MutationTrackingService.instance().retryFailedWrite(mutationId, toHost, failureReason);
}
@Override
void send()
{
RecordPointer pointer = MutationJournal.instance().lookUp(mutationId);

33
src/java/org/apache/cassandra/replication/CoordinatorLog.java
View File

@ -271,6 +271,39 @@ public abstract class CoordinatorLog
}
}
/**
* @return the computed union of remote-witnessed offsets minus local-witnessed offsets
*/
@Nullable
Offsets.Immutable collectLocallyMissingOffsets()
{
lock.readLock().lock();
try
{
Offsets.Mutable local = witnessedOffsets.get(localNodeId);
Offsets.Immutable.Builder missing = null;
for (int i = 0; i < participants.size(); i++)
{
int nodeId = participants.get(i);
if (nodeId == localNodeId) continue;
Offsets.Immutable diff = Offsets.Immutable.difference(witnessedOffsets.get(nodeId), local);
if (!diff.isEmpty())
{
if (missing == null)
{
missing = new Offsets.Immutable.Builder(logId);
}
missing.addAll(diff);
}
}
return missing != null ? missing.build() : null;
}
finally
{
lock.readLock().unlock();
}
}
Offsets.Immutable collectReconciledOffsets()
{
lock.readLock().lock();

4
src/java/org/apache/cassandra/replication/MutationSummary.java
View File

@ -25,6 +25,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.agrona.collections.Long2ObjectHashMap;
@ -274,7 +275,8 @@ public class MutationSummary
return summaries.size();
}
boolean isEmpty()
@VisibleForTesting
public boolean isEmpty()
{
return size() == 0;
}

176
src/java/org/apache/cassandra/replication/MutationTrackingService.java
View File

@ -49,6 +49,8 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.config.MutationTrackingSpec;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
@ -87,6 +89,7 @@ import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.ownership.ReplicaGroups;
import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MBeanWrapper;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.lang.String.format;
@ -96,19 +99,25 @@ import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
// TODO (expected): persistence (handle restarts)
// TODO (expected): handle topology changes
public class MutationTrackingService
public class MutationTrackingService implements MutationTrackingServiceMBean
{
public static final String MBEAN_NAME = "org.apache.cassandra.db:type=MutationTrackingService";
public static final String DISABLED_MESSAGE = "Mutation tracking is not enabled. (See mutation_tracking.enabled in cassandra.yaml)";
private static final MutationTrackingService instance;
private static final ScheduledExecutorPlus executor;
private static final MutationTrackingSpec config;
static
{
if (DatabaseDescriptor.getMutationTrackingEnabled())
config = DatabaseDescriptor.getMutationTrackingConfig();
if (config.enabled)
{
instance = new MutationTrackingService();
executor = executorFactory().scheduled("Mutation-Tracking-Service", NORMAL);
MBeanWrapper.instance.registerMBean(instance, MBEAN_NAME);
}
else
{
@ -129,12 +138,12 @@ public class MutationTrackingService
public static boolean isEnabled()
{
return DatabaseDescriptor.getMutationTrackingEnabled();
return config.enabled;
}
public static void ensureEnabled()
{
if (!DatabaseDescriptor.getMutationTrackingEnabled())
if (!config.enabled)
throw new IllegalStateException(DISABLED_MESSAGE);
}
@ -188,6 +197,7 @@ public class MutationTrackingService
private final ReplicatedOffsetsBroadcaster offsetsBroadcaster = new ReplicatedOffsetsBroadcaster();
private final LogStatePersister offsetsPersister = new LogStatePersister();
private final ActiveLogReconciler activeReconciler = new ActiveLogReconciler();
private final BackgroundReconciler backgroundReconciler = new BackgroundReconciler();
private final IncomingMutations incomingMutations = new IncomingMutations();
private final OutgoingMutations outgoingMutations = new OutgoingMutations();
@ -223,14 +233,53 @@ public class MutationTrackingService
onNewClusterMetadata(null, metadata);
if (!keyspaceShards.isEmpty() && !config.background_reconciliation_enabled)
logBackgroundReconciliationDisabledWarning(keyspaceShards.keySet());
offsetsBroadcaster.start();
offsetsPersister.start();
backgroundReconciler.start();
ExpiredStatePurger.instance.register(incomingMutations);
started = true;
}
@Override
public void setMutationTrackingBackgroundReconciliationEnabled(boolean enabled)
{
if (enabled != config.background_reconciliation_enabled)
{
logger.info("{} mutation tracking background reconciliation", enabled ? "Enabling" : "Disabling");
config.background_reconciliation_enabled = enabled;
}
}
@Override
public boolean getMutationTrackingBackgroundReconciliationEnabled()
{
return config.background_reconciliation_enabled;
}
@Override
public void setMutationTrackingBackgroundReconciliationIntervalMilliseconds(long intervalMilliseconds)
{
if (intervalMilliseconds != config.background_reconciliation_interval.toMilliseconds())
{
DurationSpec.LongMillisecondsBound backgroundReconciliationInterval =
new DurationSpec.LongMillisecondsBound(intervalMilliseconds, TimeUnit.MILLISECONDS);
logger.info("Setting mutation tracking background reconciliation interval from {} to {}",
config.background_reconciliation_interval, backgroundReconciliationInterval);
config.background_reconciliation_interval = backgroundReconciliationInterval;
}
}
@Override
public long getMutationTrackingBackgroundReconciliationIntervalMilliseconds()
{
return config.background_reconciliation_interval.toMilliseconds();
}
public void pauseOffsetBroadcast(boolean pause)
{
offsetsBroadcaster.pauseOffsetBroadcast(pause);
@ -862,6 +911,14 @@ public class MutationTrackingService
// recalculating the shards will repopulate this via the existing callbacks
log2ShardMap = new ConcurrentHashMap<>();
keyspaceShards = applyUpdatedMetadata(keyspaceShards, prev, next, this::nextLogId, this::onNewLog);
if (!config.background_reconciliation_enabled)
{
Set<String> newKeyspaces = new HashSet<>(keyspaceShards.keySet());
newKeyspaces.removeAll(originalKeyspaceShards.keySet());
if (!newKeyspaces.isEmpty())
logBackgroundReconciliationDisabledWarning(newKeyspaces);
}
}
catch (Throwable t)
{
@ -1047,6 +1104,12 @@ public class MutationTrackingService
return unwrapped;
}
private void logBackgroundReconciliationDisabledWarning(Set<String> keyspaces)
{
logger.warn("Background reconciliation is disabled but mutation tracking keyspaces exist: {}. " +
"Unreconciled mutations will not be automatically repaired in the background.", keyspaces);
}
public static class KeyspaceShards
{
private enum UpdateDecision
@ -1374,6 +1437,93 @@ public class MutationTrackingService
return rows.one().getInt("host_log_id");
}
private static class BackgroundReconciler
{
void start()
{
scheduleNext();
}
private void scheduleNext()
{
long intervalMillis = config.background_reconciliation_interval.toMilliseconds();
executor.schedule(this::runAndReschedule, intervalMillis, TimeUnit.MILLISECONDS);
}
private void runAndReschedule()
{
try
{
run();
}
finally
{
scheduleNext();
}
}
void run()
{
MutationTrackingService.instance().forEachKeyspace(this::run);
}
private void run(KeyspaceShards shards)
{
if (config.background_reconciliation_enabled)
shards.forEachShard(this::run);
}
private void run(Shard shard)
{
try
{
List<Offsets.Immutable> missing = shard.collectLocallyMissingOffsets();
if (missing.isEmpty()) return;
for (Offsets.Immutable offsets : missing)
{
// Prefer pulling from the coordinator
int coordinatorHostId = offsets.logId().hostId();
InetAddressAndPort coordinator = ClusterMetadata.current().directory.endpoint(new NodeId(coordinatorHostId));
InetAddressAndPort pullFrom = FailureDetector.instance.isAlive(coordinator)
? coordinator
: findAliveReplica(shard, coordinatorHostId);
if (pullFrom == null)
{
logger.debug("No coordinator or replica is available to process the pull mutation request for missing offset {}",
offsets);
continue; // No reachable source
}
// TODO (expected): backoff, rate limits, per host and total
PullMutationsRequest request = new PullMutationsRequest(offsets);
logger.trace("Requesting pull mutation request from replica {} for missing offset {}", pullFrom, offsets);
MessagingService.instance().send(Message.out(Verb.PULL_MUTATIONS_REQ, request), pullFrom);
}
}
catch (Throwable throwable)
{
// Avoid throwing an exception in the reconciliation step to prevent the scheduled task from
// being killed
logger.error("Exception encountered during background reconciliation of shard={}", shard, throwable);
}
}
private InetAddressAndPort findAliveReplica(Shard shard, int excludeHostId)
{
for (InetAddressAndPort replica : shard.remoteReplicas())
{
int replicaId = ClusterMetadata.current().directory.peerId(replica).id();
if (replicaId != excludeHostId && FailureDetector.instance.isAlive(replica))
{
logger.trace("Found alive replica {} with replica id {}", replica, replicaId);
return replica;
}
}
return null;
}
}
// TODO (later): a more intelligent heuristic for offsets included in broadcasts
private static class ReplicatedOffsetsBroadcaster
{
@ -1509,6 +1659,24 @@ public class MutationTrackingService
activeReconciler.resumeRegularPriorityForTesting();
}
@VisibleForTesting
public void reconcileForTesting()
{
backgroundReconciler.run();
}
@VisibleForTesting
public void pauseBackgroundReconciler()
{
config.background_reconciliation_enabled = false;
}
@VisibleForTesting
public void resumeBackgroundReconciler()
{
config.background_reconciliation_enabled = true;
}
@VisibleForTesting
public static class TestAccess
{

49
src/java/org/apache/cassandra/replication/MutationTrackingServiceMBean.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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.replication;
/**
* MBean exposing functionality for the Mutation Tracking service
*/
public interface MutationTrackingServiceMBean
{
/**
* Sets the background reconciliation state to enabled/disabled based on the {@param enabled} parameter
*
* @param enabled whether the background reconciliation is enabled or disabled for the mutation tracking service
*/
void setMutationTrackingBackgroundReconciliationEnabled(boolean enabled);
/**
* @return the state of the background reconciliation for the mutation tracking service
*/
boolean getMutationTrackingBackgroundReconciliationEnabled();
/**
* Sets the background reconciliation interval to the provided {@param intervalMilliseconds} value
*
* @param intervalMilliseconds the interval value in milliseconds
*/
void setMutationTrackingBackgroundReconciliationIntervalMilliseconds(long intervalMilliseconds);
/**
* @return the interval, in milliseconds, in which the background reconciliation runs when enabled
*/
long getMutationTrackingBackgroundReconciliationIntervalMilliseconds();
}

16
src/java/org/apache/cassandra/replication/Shard.java
View File

@ -299,6 +299,22 @@ public class Shard
return new BroadcastLogOffsets(keyspace, range, offsets, durable);
}
/**
* @return the list of the collected locally missing offsets for the logs owned by this coordinator on
* this shard
*/
List<Offsets.Immutable> collectLocallyMissingOffsets()
{
List<Offsets.Immutable> result = new ArrayList<>(logs.size());
for (CoordinatorLog log : logs.values())
{
Offsets.Immutable missing = log.collectLocallyMissingOffsets();
if (missing != null)
result.add(missing);
}
return result;
}
void collectDurablyReconciledOffsets(Log2OffsetsMap.Mutable into)
{
logs.values().forEach(log -> log.collectDurablyReconciledOffsets(into));

32
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -119,6 +119,8 @@ import org.apache.cassandra.metrics.ThreadPoolMetrics;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.MessagingServiceMBean;
import org.apache.cassandra.profiler.AsyncProfilerMBean;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.replication.MutationTrackingServiceMBean;
import org.apache.cassandra.service.ActiveRepairServiceMBean;
import org.apache.cassandra.service.AsyncProfilerService;
import org.apache.cassandra.service.AutoRepairService;
@ -190,6 +192,7 @@ public class NodeProbe implements AutoCloseable
protected PermissionsCacheMBean pcProxy;
protected RolesCacheMBean rcProxy;
protected AutoRepairServiceMBean autoRepairProxy;
protected MutationTrackingServiceMBean mutationTrackingProxy;
protected AsyncProfilerMBean asyncProfilerProxy;
protected GuardrailsMBean grProxy;
protected volatile Output output;
@ -339,6 +342,10 @@ public class NodeProbe implements AutoCloseable
name = new ObjectName(AutoRepairService.MBEAN_NAME);
autoRepairProxy = JMX.newMBeanProxy(mbeanServerConn, name, AutoRepairServiceMBean.class);
name = new ObjectName(MutationTrackingService.MBEAN_NAME);
if (mbeanServerConn.isRegistered(name))
mutationTrackingProxy = JMX.newMBeanProxy(mbeanServerConn, name, MutationTrackingServiceMBean.class);
name = new ObjectName(AsyncProfilerService.MBEAN_NAME);
asyncProfilerProxy = JMX.newMBeanProxy(mbeanServerConn, name, AsyncProfilerMBean.class);
@ -2847,6 +2854,31 @@ public class NodeProbe implements AutoCloseable
throw new IOException("Invalid keyspace or table name", e);
}
}
public boolean isMutationTrackingDisabled()
{
return mutationTrackingProxy == null;
}
public boolean getMutationTrackingBackgroundReconciliationEnabled()
{
return mutationTrackingProxy.getMutationTrackingBackgroundReconciliationEnabled();
}
public void setMutationTrackingBackgroundReconciliationEnabled(boolean enabled)
{
mutationTrackingProxy.setMutationTrackingBackgroundReconciliationEnabled(enabled);
}
public long getMutationTrackingBackgroundReconciliationIntervalMilliseconds()
{
return mutationTrackingProxy.getMutationTrackingBackgroundReconciliationIntervalMilliseconds();
}
public void setMutationTrackingBackgroundReconciliationIntervalMilliseconds(long intervalMilliseconds)
{
mutationTrackingProxy.setMutationTrackingBackgroundReconciliationIntervalMilliseconds(intervalMilliseconds);
}
}
class ColumnFamilyStoreMBeanIterator implements Iterator<Map.Entry<String, ColumnFamilyStoreMBean>>

112
src/java/org/apache/cassandra/tools/nodetool/MTAdmin.java Normal file
View File

@ -0,0 +1,112 @@
/*
* 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.tools.nodetool;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.List;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.tools.NodeProbe;
import picocli.CommandLine.Command;
import picocli.CommandLine.Parameters;
import static com.google.common.base.Preconditions.checkArgument;
@Command(name = "mtadmin",
description = "Manage mutation tracking",
subcommands = { MTAdmin.GetConfig.class,
MTAdmin.SetConfig.class })
public class MTAdmin extends AbstractCommand
{
@Override
protected void execute(NodeProbe probe)
{
AbstractCommand cmd = new GetConfig();
cmd.probe(probe);
cmd.logger(output);
cmd.run();
}
@Command(name = "getconfig", description = "Print mutation tracking configurations")
public static class GetConfig extends AbstractCommand
{
@VisibleForTesting
protected PrintStream out = System.out;
@Override
public void execute(NodeProbe probe)
{
if (probe.isMutationTrackingDisabled())
{
out.println("Mutation tracking is not enabled");
return;
}
out.println("background_reconciliation_enabled: " + probe.getMutationTrackingBackgroundReconciliationEnabled());
out.println("background_reconciliation_interval_ms: " + probe.getMutationTrackingBackgroundReconciliationIntervalMilliseconds());
}
}
@Command(name = "setconfig", description = "Sets the mutation tracking configuration")
public static class SetConfig extends AbstractCommand
{
@VisibleForTesting
protected List<String> args = new ArrayList<>();
@Parameters(index = "0", arity = "0..1", description = { "Mutation tracking param type.",
"Possible parameters: " +
"[background_reconciliation_enabled|background_reconciliation_interval_ms]" })
public String paramType;
@Parameters(index = "1", description = "Mutation tracking param value", arity = "0..1")
public String paramValue;
@VisibleForTesting
protected PrintStream out = System.out;
@Override
public void execute(NodeProbe probe)
{
args = args.isEmpty() ? CommandUtils.concatArgs(paramType, paramValue) : args;
checkArgument(args.size() == 2, "mtadmin setconfig requires param-type and value args.");
String type = args.get(0);
String value = args.get(1);
if (probe.isMutationTrackingDisabled())
{
out.println("Mutation tracking is not enabled");
return;
}
switch (type)
{
case "background_reconciliation_enabled":
probe.setMutationTrackingBackgroundReconciliationEnabled(Boolean.parseBoolean(value));
break;
case "background_reconciliation_interval_ms":
probe.setMutationTrackingBackgroundReconciliationIntervalMilliseconds(Long.parseLong(value));
break;
default:
throw new IllegalArgumentException("Unknown parameter: " + type);
}
}
}
}

1
src/java/org/apache/cassandra/tools/nodetool/NodetoolCommand.java
View File

@ -147,6 +147,7 @@ import static org.apache.cassandra.tools.nodetool.Help.printTopCommandUsage;
ListSnapshots.class,
Move.Abort.class,
Move.class,
MTAdmin.class,
NetStats.class,
PauseHandoff.class,
ProfileLoad.class,

6
test/distributed/org/apache/cassandra/distributed/test/ReadRepairQueryTester.java
View File

@ -120,9 +120,11 @@ public abstract class ReadRepairQueryTester extends TestBaseImpl
@BeforeClass
public static void setupCluster() throws IOException
{
cluster = init(Cluster.build(NUM_NODES)
// Disable background reconciler for read repair tests to avoid having the
// background reconciler repair before the read repair takes effect
cluster = init(disableBackgroundReconciler(Cluster.build(NUM_NODES)
.withConfig(config -> config.set("read_request_timeout", "1m")
.set("write_request_timeout", "1m"))
.set("write_request_timeout", "1m")))
.start());
}

26
test/distributed/org/apache/cassandra/distributed/test/ReadRepairTestBase.java
View File

@ -139,7 +139,8 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
private void testReadRepair(ReadRepairStrategy strategy, boolean brrThroughAccord) throws Throwable
{
try (Cluster cluster = init(Cluster.create(3, c -> c.with(Feature.GOSSIP, Feature.NETWORK))))
try (Cluster cluster = init(Cluster.create(3, c -> c.with(Feature.GOSSIP, Feature.NETWORK)
.set("mutation_tracking.background_reconciliation_enabled", "false"))))
{
createKeyspace(cluster);
TransactionalMode transactionalMode = brrThroughAccord ? TransactionalMode.test_unsafe_writes : TransactionalMode.off;
@ -175,7 +176,9 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
@Test
public void readRepairTimeoutTest() throws Throwable
{
try (Cluster cluster = init(Cluster.create(3, c -> c.with(Feature.GOSSIP, Feature.NETWORK)))) {
try (Cluster cluster = init(Cluster.create(3, c -> c.with(Feature.GOSSIP, Feature.NETWORK)
.set("mutation_tracking.background_reconciliation_enabled", "false"))))
{
final long reducedReadTimeout = 3000L;
createKeyspace(cluster);
cluster.forEach(i -> i.runOnInstance(() -> DatabaseDescriptor.setReadRpcTimeout(reducedReadTimeout)));
@ -209,7 +212,7 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
@Test
public void failingReadRepairTest() throws Throwable
{
try (Cluster cluster = init(builder().withNodes(3).start()))
try (Cluster cluster = init(disableBackgroundReconciler(builder().withNodes(3)).start()))
{
createKeyspace(cluster);
cluster.schemaChange(withTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH read_repair='blocking'"));
@ -237,7 +240,7 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
MutationTrackingUtils.fixmeSkipIfTracked(replicationType(), "Token moves not supported");
// TODO: rewrite using FuzzTestBase to control progress through decommission
// TODO: fails with vnode enabled
try (Cluster cluster = init(Cluster.build(4).withoutVNodes().start(), 3))
try (Cluster cluster = init(disableBackgroundReconciler(Cluster.build(4).withoutVNodes()).start(), 3))
{
List<Token> tokens = cluster.tokens();
@ -296,7 +299,7 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
public void alterRFAndRunReadRepair() throws Throwable
{
MutationTrackingUtils.fixmeSkipIfTracked(replicationType(), "RF changes not supported");
try (Cluster cluster = builder().withNodes(2).start())
try (Cluster cluster = disableBackgroundReconciler(builder().withNodes(2)).start())
{
cluster.schemaChange(format("CREATE KEYSPACE %s WITH replication = " +
"{'class': 'SimpleStrategy', 'replication_factor': 1} " +
@ -358,7 +361,7 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
*/
private void testRangeSliceQueryWithTombstones(boolean flush) throws Throwable
{
try (Cluster cluster = init(Cluster.create(2)))
try (Cluster cluster = init(Cluster.create(2, c -> c.set("mutation_tracking.background_reconciliation_enabled", "false"))))
{
createKeyspace(cluster);
cluster.schemaChange(withTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY(k, c))"));
@ -425,14 +428,15 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
MutationTrackingUtils.fixmeSkipIfTracked(replicationType(), "Token moves not supported");
ExecutorPlus es = ExecutorFactory.Global.executorFactory().sequential("query-executor");
String key = "test1";
try (Cluster cluster = init(Cluster.build()
try (Cluster cluster = init(disableBackgroundReconciler(Cluster.build()
.withConfig(config -> config.with(Feature.GOSSIP, Feature.NETWORK)
.set("read_request_timeout", format("%dms", Integer.MAX_VALUE))
.set("native_transport_timeout", format("%dms", Integer.MAX_VALUE))
.set("mutation_tracking.background_reconciliation_enabled", "false")
)
.withTokenSupplier(TokenSupplier.evenlyDistributedTokens(4))
.withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
.withNodes(3)
.withNodes(3))
.start()))
{
createKeyspace(cluster);
@ -525,7 +529,7 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
@Test
public void testGCableTombstoneResurrectionOnRangeSliceQuery() throws Throwable
{
try (Cluster cluster = init(Cluster.create(2)))
try (Cluster cluster = init(Cluster.create(2, c -> c.set("mutation_tracking.background_reconciliation_enabled", "false"))))
{
createKeyspace(cluster);
cluster.schemaChange(withTable("CREATE TABLE %s (k int, c int, PRIMARY KEY(k, c)) " +
@ -563,9 +567,9 @@ public abstract class ReadRepairTestBase extends TestBaseImpl
@Test
public void partitionDeletionRTTimestampTieTest() throws Throwable
{
try (Cluster cluster = init(builder()
try (Cluster cluster = init(disableBackgroundReconciler(builder()
.withNodes(3)
.withInstanceInitializer(RRHelper::install)
.withInstanceInitializer(RRHelper::install))
.start()))
{
createKeyspace(cluster);

31
test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
View File

@ -83,6 +83,7 @@ import org.apache.cassandra.distributed.shared.DistributedTestBase;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.service.accord.AccordCache;
import static java.lang.System.currentTimeMillis;
@ -452,4 +453,34 @@ public class TestBaseImpl extends DistributedTestBase
}
}
}
/**
* Useful for tests that rely on different repair mechanisms, for example for read repair tests
* where the test relies on the read repair machinery, we want to explicitly disable the
* background reconciliation process.
*
* @param cluster the cluster for the test
* @return the cluster with the background reconciliation process disabled on all instances
*/
public static Cluster disableBackgroundReconciler(Cluster cluster)
{
for (IInvokableInstance instance : cluster)
{
instance.runOnInstance(() -> MutationTrackingService.instance().pauseBackgroundReconciler());
}
return cluster;
}
/**
* Useful for tests that rely on different repair mechanisms, for example for read repair tests
* where the test relies on the read repair machinery, we want to start the cluster with the
* background reconciliation process disabled.
*
* @param builder the cluster builder
* @return the cluster builder with the background reconciliation process configuration disabled
*/
public static Cluster.Builder disableBackgroundReconciler(Cluster.Builder builder)
{
return builder.appendConfig(c -> c.set("mutation_tracking.background_reconciliation_enabled", "false"));
}
}

3
test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkBase.java
View File

@ -81,7 +81,8 @@ public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest
.set("read_request_timeout", "180s")
.set("write_request_timeout", "180s")
.set("native_transport_timeout", "180s")
.set("slow_query_log_timeout", "180s");
.set("slow_query_log_timeout", "180s")
.set("mutation_tracking.background_reconciliation_enabled", "false");
}
@Override

3
test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java
View File

@ -158,7 +158,8 @@ public class StatefulASTBase extends TestBaseImpl
protected void clusterConfig(IInstanceConfig config)
{
config.set("repair.retries.max_attempts", Integer.MAX_VALUE);
config.set("repair.retries.max_attempts", Integer.MAX_VALUE)
.set("mutation_tracking.background_reconciliation_enabled", "false");
}
protected void clusterInitializer(ClassLoader cl, int node)

2
test/distributed/org/apache/cassandra/distributed/test/sai/PartialUpdateHandlingTest.java
View File

@ -89,7 +89,7 @@ public class PartialUpdateHandlingTest extends TestBaseImpl
@BeforeClass
public static void setUpCluster() throws IOException
{
CLUSTER = Cluster.build(NODES).withConfig(config -> config.set("hinted_handoff_enabled", false).with(GOSSIP).with(NETWORK)).start();
CLUSTER = disableBackgroundReconciler(Cluster.build(NODES).withConfig(config -> config.set("hinted_handoff_enabled", false).with(GOSSIP).with(NETWORK))).start();
for (ReplicationType replicationType : ReplicationType.values())
{

165
test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingTest.java
View File

@ -19,6 +19,7 @@
package org.apache.cassandra.distributed.test.tracking;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Assert;
import org.junit.Ignore;
@ -49,9 +50,15 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.distributed.test.tracking.MutationTrackingReadReconciliationTest.awaitNodeAlive;
import static org.apache.cassandra.distributed.test.tracking.MutationTrackingReadReconciliationTest.awaitNodeDead;
import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.assertMatchingSummaryIdSpaceForKey;
import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.getOnlyLogId;
import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.numLogReconciliations;
import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.summaryForKey;
import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.summaryIdSpace;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
// TODO This test would be a lot faster if it had a shared cluster
@ -436,4 +443,162 @@ public class MutationTrackingTest extends TestBaseImpl
}), 10);
}
}
@Test
public void testBackgroundPullReconciliation() throws Throwable
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
.set("write_request_timeout", "1000ms"))
.start())
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " +
"{'class': 'SimpleStrategy', 'replication_factor': 3} " +
"AND replication_type='tracked'"));
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (k int PRIMARY KEY, v int)"));
// 1. Partition node 3 and pause push-side retries on all nodes
cluster.filters().allVerbs().to(3).drop();
cluster.filters().allVerbs().from(3).drop();
for (int i = 1; i <= 2; i++)
cluster.get(i).runOnInstance(() -> Gossiper.instance.convict(InetAddressAndPort.getByNameUnchecked("127.0.0.3"), Double.MAX_VALUE));
for (int i = 1; i <= 3; i++)
cluster.get(i).runOnInstance(() -> {
MutationTrackingService.instance().pauseActiveReconciler();
MutationTrackingService.instance().pauseBackgroundReconciler();
});
// wait until node 1 marks node 3 as dead
awaitNodeDead(cluster.get(1), cluster.get(3));
// 2. Write at QUORUM - succeeds on nodes 1, 2 but node 3 won't get the write
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (k, v) VALUES (1, 1)"),
ConsistencyLevel.QUORUM);
// Sleep one second until write timeout elapses
TimeUnit.SECONDS.sleep(1);
// 3. Capture expected state from node 1
MutationSummary expected = summaryForKey(cluster.get(1), KEYSPACE, "tbl", /* key */1);
// 4. Ensure node 3 does NOT have the mutation yet
cluster.get(3).runOnInstance(() -> {
TableMetadata table = Schema.instance.getTableMetadata(KEYSPACE, "tbl");
assertNotNull(table);
DecoratedKey dk = Murmur3Partitioner.instance.decorateKey(ByteBufferUtil.bytes(1));
MutationSummary summary = MutationTrackingService.instance()
.createSummaryForKey(dk, table.id, false);
assertTrue("Node 3 should have no mutations yet", summary.isEmpty());
});
// 5. Reset state for node 3, let broadcasts propagate
cluster.filters().reset();
awaitNodeAlive(cluster.get(1), cluster.get(3));
awaitNodeAlive(cluster.get(3), cluster.get(1));
// Now broadcast offsets so node 3 learns what nodes 1 and 2 have
for (int i = 1; i <= 3; i++)
cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting());
cluster.get(1).runOnInstance(() -> MutationTrackingService.instance().resumeActiveReconciler());
// 6. Trigger the background reconciler on node 3 ONLY (no reads, no push retries)
cluster.get(3).runOnInstance(() -> {
MutationTrackingService.instance().resumeBackgroundReconciler();
MutationTrackingService.instance().reconcileForTesting();
});
// 7. Wait for the pull request to be processed and mutation to arrive
TimeUnit.SECONDS.sleep(2);
// Broadcast again so reconciliation state converges
for (int i = 1; i <= 3; i++)
cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting());
// 8. Verify node 3 now has the mutation (pulled via background reconciler)
assertMatchingSummaryIdSpaceForKey(cluster.get(3), KEYSPACE, "tbl", /* key */1, expected);
// 9. Verify no read reconciliation was triggered
assertEquals(0, numLogReconciliations(cluster.get(3)));
}
}
@Test
public void testBackgroundPullReconciliationWhenCoordinatorDown() throws Throwable
{
try (Cluster cluster = Cluster.build(3).withConfig(cfg -> cfg.with(Feature.NETWORK)
.with(Feature.GOSSIP)
.set("write_request_timeout", "1000ms"))
.start())
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " +
"{'class': 'SimpleStrategy', 'replication_factor': 3} " +
"AND replication_type='tracked'"));
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (k int PRIMARY KEY, v int)"));
// 1. Pause push-side retires and background reconciler on all nodes
for (int i = 1; i <= 3; i++)
cluster.get(i).runOnInstance(() -> {
MutationTrackingService.instance().pauseActiveReconciler();
MutationTrackingService.instance().pauseBackgroundReconciler();
});
// 2. Partition node 3, then write at QUORUM from coordinator (node 1)
cluster.filters().allVerbs().to(3).drop();
cluster.filters().allVerbs().from(3).drop();
awaitNodeDead(cluster.get(1), cluster.get(3));
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (k, v) VALUES (1, 1)"),
ConsistencyLevel.QUORUM);
// Sleep one second until write timeout elapses
TimeUnit.SECONDS.sleep(1);
// 3. Capture expected state from node 1 (before we partition it)
MutationSummary expected = summaryForKey(cluster.get(1), KEYSPACE, "tbl", 1);
// 4. Heal node 3, then partition node 1 (the coordinator)
cluster.filters().reset();
awaitNodeAlive(cluster.get(2), cluster.get(3));
awaitNodeAlive(cluster.get(3), cluster.get(2));
cluster.filters().allVerbs().to(1).drop();
cluster.filters().allVerbs().from(1).drop();
for (int i = 2; i <= 3; i++)
cluster.get(i).runOnInstance(() -> Gossiper.instance.convict(InetAddressAndPort.getByNameUnchecked("127.0.0.1"), Double.MAX_VALUE));
awaitNodeDead(cluster.get(3), cluster.get(1));
awaitNodeDead(cluster.get(2), cluster.get(1));
// 5. Broadcast offsets between nodes 2 and 3 only (Node 3 learns what node 2 has witnessed
// should discover the gap)
for (int i = 2; i <= 3; i++)
cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting());
// 6. Trigger background reconciler on node 3. Coordinator is down, so it is expected to fallback to node 2
// (the remaining alive replica)
cluster.get(3).runOnInstance(() -> {
MutationTrackingService.instance().resumeBackgroundReconciler();
MutationTrackingService.instance().reconcileForTesting();
});
// 7. Resume the active reconciler on instance 2, so whenever instance 2 receives the request
// it will process the PULL_MUTATIONS_REQ verb
cluster.get(2).runOnInstance(() -> MutationTrackingService.instance().resumeActiveReconciler());
// Wait for the pull request to be processed and mutation to arrive
TimeUnit.SECONDS.sleep(2);
// Broadcast again so reconciliation state converges
for (int i = 2; i <= 3; i++)
cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().broadcastOffsetsForTesting());
// 8. Verify node 3 pulled the mutation from node 2 (the fallback replica)
assertMatchingSummaryIdSpaceForKey(cluster.get(3), KEYSPACE, "tbl", 1, expected);
// 9. Verify no read reconciliation was involved
assertEquals(0, numLogReconciliations(cluster.get(3)));
}
}
}

6
test/distributed/org/apache/cassandra/distributed/test/tracking/OffsetBroadcastTest.java
View File

@ -43,9 +43,9 @@ public class OffsetBroadcastTest extends TestBaseImpl
@Test
public void testBroadcastOffsets() throws Throwable
{
try (Cluster cluster = Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK).with(Feature.GOSSIP))
.start())
try (Cluster cluster = disableBackgroundReconciler(Cluster.build(3)
.withConfig(cfg -> cfg.with(Feature.NETWORK).with(Feature.GOSSIP)))
.start())
{
cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " +

2
test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java
View File

@ -163,7 +163,7 @@ public class TrackedImportFailureTest extends TrackedTransferTestBase
{
ActivationRequest.Phase phase = COMMIT;
int MISSED_ACTIVATION = 2;
try (Cluster cluster = cluster(TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION)))
try (Cluster cluster = disableBackgroundReconciler(cluster(TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION))))
{
TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.setup(cluster, phase);
createSchema(cluster);

2
test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedNonZeroCopyRepairTransferTest.java
View File

@ -27,6 +27,6 @@ public class TrackedNonZeroCopyRepairTransferTest extends TrackedRepairTransferS
@BeforeClass
public static void setup() throws IOException
{
cluster = cluster(NON_ZCS_CONFIG);
cluster = disableBackgroundReconciler(cluster(NON_ZCS_CONFIG));
}
}

6
test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedRepairFailureTest.java
View File

@ -65,7 +65,7 @@ public class TrackedRepairFailureTest extends TrackedRepairTransferTestBase
@Test
public void testFullRepairPartiallyCompleteAnomaly() throws IOException, ExecutionException, InterruptedException, TimeoutException
{
try (Cluster cluster = cluster(StreamReceiverFailureHelper::install))
try (Cluster cluster = disableBackgroundReconciler(cluster(StreamReceiverFailureHelper::install)))
{
cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3} AND replication_type='tracked';");
String TABLE_SCHEMA_CQL = "CREATE TABLE " + tableWithKeyspace(KEYSPACE) + " (k INT PRIMARY KEY, v INT)";
@ -192,7 +192,7 @@ public class TrackedRepairFailureTest extends TrackedRepairTransferTestBase
@Test
public void testFullRepairCleanupOnFailure() throws IOException, ExecutionException, InterruptedException, TimeoutException
{
try (Cluster cluster = cluster(StreamReceiverFailureHelper::install))
try (Cluster cluster = disableBackgroundReconciler(cluster(StreamReceiverFailureHelper::install)))
{
cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3} AND replication_type='tracked';");
String TABLE_SCHEMA_CQL = "CREATE TABLE " + KEYSPACE + '.' + TABLE + " (k INT PRIMARY KEY, v INT)";
@ -259,7 +259,7 @@ public class TrackedRepairFailureTest extends TrackedRepairTransferTestBase
@Test
public void testRepairFailsOnMissedActivation() throws IOException
{
try (Cluster cluster = cluster(ByteBuddyInjections.SkipActivation.install(2, 3)))
try (Cluster cluster = disableBackgroundReconciler(cluster(ByteBuddyInjections.SkipActivation.install(2, 3))))
{
cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3} AND replication_type='tracked';");
cluster.schemaChange("CREATE TABLE " + tableWithKeyspace(KEYSPACE) + " (k INT PRIMARY KEY, v INT)");

6
test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedTransferTestBase.java
View File

@ -88,8 +88,10 @@ public abstract class TrackedTransferTestBase extends TestBaseImpl
.set("repair_request_timeout", "2s")
.set("stream_transfer_task_timeout", "10s");
protected static final Consumer<IInstanceConfig> ZCS_CONFIG = CONFIG.andThen(cfg -> cfg.set("stream_entire_sstables", true));
protected static final Consumer<IInstanceConfig> NON_ZCS_CONFIG = CONFIG.andThen(cfg -> cfg.set("stream_entire_sstables", false));
protected static final Consumer<IInstanceConfig> ZCS_CONFIG = CONFIG.andThen(cfg -> cfg.set("stream_entire_sstables", true)
.set("mutation_tracking.background_reconciliation_enabled", false));
protected static final Consumer<IInstanceConfig> NON_ZCS_CONFIG = CONFIG.andThen(cfg -> cfg.set("stream_entire_sstables", false)
.set("mutation_tracking.background_reconciliation_enabled", false));
protected static final IIsolatedExecutor.SerializableConsumer<SSTableReader> TRANSFERS_EXIST = sstable -> {
Assertions.assertThat(sstable.getCoordinatorLogOffsets().transfers()).isNotEmpty();

2
test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedZeroCopyRepairTransferTest.java
View File

@ -27,6 +27,6 @@ public class TrackedZeroCopyRepairTransferTest extends TrackedRepairTransferSucc
@BeforeClass
public static void setup() throws IOException
{
cluster = cluster(ZCS_CONFIG);
cluster = disableBackgroundReconciler(cluster(ZCS_CONFIG));
}
}

117
test/unit/org/apache/cassandra/tools/nodetool/MTAdminTest.java Normal file
View File

@ -0,0 +1,117 @@
/*
* 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.tools.nodetool;
import java.io.PrintStream;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.apache.cassandra.tools.NodeProbe;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class MTAdminTest
{
@Mock
private NodeProbe probe;
@Mock
private PrintStream out;
private MTAdmin.SetConfig cmd;
@Before
public void setUp()
{
MockitoAnnotations.initMocks(this);
when(probe.isMutationTrackingDisabled()).thenReturn(false);
cmd = new MTAdmin.SetConfig();
cmd.out = out;
}
@Test(expected = IllegalArgumentException.class)
public void testNoArgs()
{
cmd.execute(probe);
}
@Test
public void testMutationTrackingDisabled()
{
when(probe.isMutationTrackingDisabled()).thenReturn(true);
cmd.args = List.of("background_reconciliation_enabled", "true");
cmd.execute(probe);
verify(out, times(1)).println("Mutation tracking is not enabled");
verify(probe, times(0)).setMutationTrackingBackgroundReconciliationEnabled(true);
}
@Test
public void testSetBackgroundReconciliationEnabled()
{
cmd.args = List.of("background_reconciliation_enabled", "true");
cmd.execute(probe);
verify(probe, times(1)).setMutationTrackingBackgroundReconciliationEnabled(true);
}
@Test
public void testSetBackgroundReconciliationDisabled()
{
cmd.args = List.of("background_reconciliation_enabled", "false");
cmd.execute(probe);
verify(probe, times(1)).setMutationTrackingBackgroundReconciliationEnabled(false);
}
@Test
public void testSetBackgroundReconciliationIntervalMs()
{
cmd.args = List.of("background_reconciliation_interval_ms", "5000");
cmd.execute(probe);
verify(probe, times(1)).setMutationTrackingBackgroundReconciliationIntervalMilliseconds(5000L);
}
@Test(expected = IllegalArgumentException.class)
public void testInvalidParamType()
{
cmd.args = List.of("unknown_param", "value");
cmd.execute(probe);
}
@Test(expected = NumberFormatException.class)
public void testInvalidIntervalValue()
{
cmd.args = List.of("background_reconciliation_interval_ms", "not_a_number");
cmd.execute(probe);
}
}