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

Fix additional live migration/interop merge issues 

s fc83325fa9 Fix AccordObjectSizes empty sizes
s 44bce6a08c Fix error handling when there are no column families to repair
s f1459540e5 Fix broken nowInSec deserialization
s 91c1befd1c Fix Mutation serializedSize
s b11467c200 Using simulated clock instead of raw nanoTime
s 6412b35924 Accord repair needs to use sentinel tokens
s d19c01ff01 Enable Accord repair with HappyPathFuzzTest
s 2da87f91b8 Implement abort in AccordRepairJob
s e12877c4fa move barrier to accord spec

Patch by Ariel Weisberg; Reviewed by David Capwell for CASSANDRA-19023
This commit is contained in:
Ariel Weisberg 2023-11-29 17:12:46 -05:00 committed by David Capwell
parent d07a36106d
commit a9cb71833a
18 changed files with 330 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit d9ef555302f8774ed03325ba22d38ee0b80130a8
Subproject commit 3ca9e55024192e9b7c38ad5330229830343b74f2

15
src/java/org/apache/cassandra/config/AccordSpec.java
View File

@ -27,4 +27,19 @@ public class AccordSpec
public volatile OptionaldPositiveInt shard_count = OptionaldPositiveInt.UNDEFINED;
public volatile DurationSpec.IntSecondsBound progress_log_schedule_delay = new DurationSpec.IntSecondsBound(1);
/**
* When a barrier transaction is requested how many times to repeat attempting the barrier before giving up
*/
public int barrier_retry_attempts = 5;
/**
* When a barrier transaction fails how long the initial backoff should be before being increased
* as part of exponential backoff on each attempt
*/
public DurationSpec.IntMillisecondsBound barrier_retry_inital_backoff_millis = new DurationSpec.IntMillisecondsBound("1s");
public DurationSpec.IntMillisecondsBound barrier_max_backoff = new DurationSpec.IntMillisecondsBound("10m");
public DurationSpec.IntMillisecondsBound range_barrier_timeout = new DurationSpec.IntMillisecondsBound("2m");
}

1
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -37,6 +37,7 @@ import static org.apache.cassandra.utils.LocalizeString.toUpperCaseLocalized;
/** A class that extracts system properties for the cassandra node it runs within. */
public enum CassandraRelevantProperties
{
ACCORD_REPAIR_RANGE_STEP_UPDATE_INTERVAL("cassandra.accord.repair.range_step_update_interval", "100"),
ACQUIRE_RETRY_SECONDS("cassandra.acquire_retry_seconds", "60"),
ACQUIRE_SLEEP_MS("cassandra.acquire_sleep_ms", "1000"),
ALLOCATE_TOKENS_FOR_KEYSPACE("cassandra.allocate_tokens_for_keyspace"),

15
src/java/org/apache/cassandra/config/Config.java
View File

@ -1170,21 +1170,6 @@ public class Config
public LWTStrategy lwt_strategy = LWTStrategy.migration;
public NonSerialWriteStrategy non_serial_write_strategy = NonSerialWriteStrategy.normal;
/**
* When a barrier transaction is requested how many times to repeat attempting the barrier before giving up
*/
public int accord_barrier_retry_attempts = 5;
/**
* When a barrier transaction fails how long the initial backoff should be before being increased
* as part of exponential backoff on each attempt
*/
public DurationSpec.IntMillisecondsBound accord_barrier_retry_inital_backoff_millis = new DurationSpec.IntMillisecondsBound("1s");
public DurationSpec.IntMillisecondsBound accord_barrier_max_backoff = new DurationSpec.IntMillisecondsBound("10m");
public DurationSpec.IntMillisecondsBound accord_range_barrier_timeout = new DurationSpec.IntMillisecondsBound("2m");
public volatile int max_top_size_partition_count = 10;
public volatile int max_top_tombstone_partition_count = 10;
public volatile DataStorageSpec.LongBytesBound min_tracked_partition_size = new DataStorageSpec.LongBytesBound("1MiB");

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

@ -3676,22 +3676,22 @@ public class DatabaseDescriptor
public static int getAccordBarrierRetryAttempts()
{
return conf.accord_barrier_retry_attempts;
return conf.accord.barrier_retry_attempts;
}
public static long getAccordBarrierRetryInitialBackoffMillis()
{
return conf.accord_barrier_retry_inital_backoff_millis.toMilliseconds();
return conf.accord.barrier_retry_inital_backoff_millis.toMilliseconds();
}
public static long getAccordBarrierRetryMaxBackoffMillis()
{
return conf.accord_barrier_max_backoff.toMilliseconds();
return conf.accord.barrier_max_backoff.toMilliseconds();
}
public static long getAccordRangeBarrierTimeoutNanos()
{
return conf.accord_range_barrier_timeout.to(TimeUnit.NANOSECONDS);
return conf.accord.range_barrier_timeout.to(TimeUnit.NANOSECONDS);
}
public static void setNativeTransportMaxRequestDataInFlightPerIpInBytes(long maxRequestDataInFlightInBytes)

26
src/java/org/apache/cassandra/db/virtual/LocalRepairTables.java
View File

@ -136,7 +136,7 @@ public class LocalRepairTables
{
result.row(state.id);
addState(result, state);
result.column("type", getType(state));
result.column("type", state.getType());
result.column("keyspace_name", state.keyspace);
result.column("command_id", state.cmd);
@ -169,30 +169,6 @@ public class LocalRepairTables
ranges = state.getCommonRanges();
result.column("unfiltered_ranges", ranges == null ? null : ranges.stream().map(c -> c.ranges).map(LocalRepairTables::toStringList).collect(Collectors.toList()));
}
private String getType(CoordinatorState state)
{
if (state.options.isPreview())
{
switch (state.options.getPreviewKind())
{
case ALL: return "preview full";
case REPAIRED: return "preview repaired";
case UNREPAIRED: return "preview unrepaired";
case NONE: throw new AssertionError("NONE preview kind not expected when preview repair is set");
default: throw new AssertionError("Unknown preview kind: " + state.options.getPreviewKind());
}
}
else if (state.options.accordRepair())
{
return "accord repair";
}
else if (state.options.isIncremental())
{
return "incremental";
}
return "full";
}
}
private static final class SessionTable extends AbstractVirtualTable

2
src/java/org/apache/cassandra/repair/AbstractRepairJob.java
View File

@ -28,7 +28,7 @@ import org.apache.cassandra.utils.concurrent.AsyncFuture;
public abstract class AbstractRepairJob extends AsyncFuture<RepairResult> implements Runnable
{
private final SharedContext ctx;
protected final SharedContext ctx;
public final JobState state;
protected final RepairJobDesc desc;
protected final RepairSession session;

43
src/java/org/apache/cassandra/repair/AccordRepairJob.java
View File

@ -19,27 +19,27 @@
package org.apache.cassandra.repair;
import java.math.BigInteger;
import java.util.List;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.BarrierType;
import accord.api.RoutingKey;
import accord.primitives.Ranges;
import accord.primitives.Seekables;
import org.apache.cassandra.dht.AccordSplitter;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.AccordTopologyUtils;
import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
import org.apache.cassandra.service.consensus.migration.ConsensusTableMigrationState.ConsensusMigrationRepairResult;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.Epoch;
import static com.google.common.base.Preconditions.checkState;
import static java.util.Collections.emptyList;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.config.CassandraRelevantProperties.ACCORD_REPAIR_RANGE_STEP_UPDATE_INTERVAL;
/*
* Accord repair consists of creating a barrier transaction for all the ranges which ensure that all Accord transactions
@ -47,6 +47,8 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime;
*/
public class AccordRepairJob extends AbstractRepairJob
{
private static final Logger logger = LoggerFactory.getLogger(AccordRepairJob.class);
public static final BigInteger TWO = BigInteger.valueOf(2);
private final Ranges ranges;
@ -57,16 +59,14 @@ public class AccordRepairJob extends AbstractRepairJob
private Epoch minEpoch = ClusterMetadata.current().epoch;
private volatile Throwable shouldAbort = null;
public AccordRepairJob(RepairSession repairSession, String cfname)
{
super(repairSession, cfname);
List<Range<Token>> normalizedRanges = Range.normalize(desc.ranges);
IPartitioner partitioner = normalizedRanges.get(0).left.getPartitioner();
TokenRange[] tokenRanges = new TokenRange[normalizedRanges.size()];
for (int i = 0; i < normalizedRanges.size(); i++)
tokenRanges[i] = new TokenRange(new TokenKey(ks.getName(), normalizedRanges.get(i).left), new TokenKey(ks.getName(), normalizedRanges.get(i).right));
this.ranges = Ranges.of(tokenRanges);
this.splitter = partitioner.accordSplitter().apply(Ranges.of(tokenRanges));
IPartitioner partitioner = desc.ranges.iterator().next().left.getPartitioner();
this.ranges = AccordTopologyUtils.toAccordRanges(desc.keyspace, desc.ranges);
this.splitter = partitioner.accordSplitter().apply(ranges);
}
@Override
@ -91,15 +91,19 @@ public class AccordRepairJob extends AbstractRepairJob
@Override
void abort(@Nullable Throwable reason)
{
throw new UnsupportedOperationException("Have not implemented this yet, and the job runs synchronously so it isn't abortable");
shouldAbort = reason == null ? new RuntimeException("Abort") : reason;
}
private void repairRange(TokenRange range)
private void repairRange(TokenRange range) throws Throwable
{
int rangeStepUpdateInterval = ACCORD_REPAIR_RANGE_STEP_UPDATE_INTERVAL.getInt();
RoutingKey remainingStart = range.start();
BigInteger rangeSize = splitter.sizeOf(range);
if (rangeStep == null)
rangeStep = BigInteger.ONE.max(splitter.divide(rangeSize, 1000));
{
BigInteger divide = splitter.divide(rangeSize, 1000);
rangeStep = divide.equals(BigInteger.ZERO) ? rangeSize : BigInteger.ONE.max(divide);
}
BigInteger offset = BigInteger.ZERO;
@ -107,14 +111,16 @@ public class AccordRepairJob extends AbstractRepairJob
int iteration = 0;
while (true)
{
if (shouldAbort != null)
throw shouldAbort;
iteration++;
if (iteration % 100 == 0)
if (iteration % rangeStepUpdateInterval == 0)
rangeStep = rangeStep.multiply(TWO);
BigInteger remaining = rangeSize.subtract(offset);
BigInteger length = remaining.min(rangeStep);
long start = nanoTime();
long start = ctx.clock().nanoTime();
boolean dependencyOverflow = false;
try
{
@ -123,7 +129,10 @@ public class AccordRepairJob extends AbstractRepairJob
if (splitter.compare(offset, rangeSize) >= 0)
{
if (remainingStart.equals(range.end()))
{
logger.info("Completed barriers for {} in {} iterations", range, iteration - 1);
return;
}
// Final repair is whatever remains
toRepair = range.newRange(remainingStart, range.end());

1
src/java/org/apache/cassandra/repair/RepairCoordinator.java
View File

@ -301,6 +301,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
this.traceState = maybeCreateTraceState(columnFamilies);
notifyStarting();
NeighborsAndRanges neighborsAndRanges = getNeighborsAndRanges();
// We test to validate the start JMX notification is seen before we compute neighbors and ranges
// but in state (vtable) tracking, we rely on getNeighborsAndRanges to know where we are running repair...
// JMX start != state start, its possible we fail in getNeighborsAndRanges and state start is never reached

10
src/java/org/apache/cassandra/repair/messages/RepairOption.java
View File

@ -350,6 +350,16 @@ public class RepairOption
this.accordRepair = accordRepair;
}
public RepairOption withAccordRepair(boolean accordRepair)
{
RepairOption repairOption = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges, pullRepair, forceRepair, previewKind, optimiseStreams, ignoreUnreplicatedKeyspaces, repairPaxos, paxosOnly, dontPurgeTombstones, accordRepair);
repairOption.columnFamilies.addAll(columnFamilies);
repairOption.dataCenters.addAll(dataCenters);
repairOption.hosts.addAll(hosts);
repairOption.ranges.addAll(ranges);
return repairOption;
}
public RepairParallelism getParallelism()
{
return parallelism;

24
src/java/org/apache/cassandra/repair/state/CoordinatorState.java
View File

@ -64,6 +64,30 @@ public class CoordinatorState extends AbstractState<CoordinatorState.State, Time
this.options = Objects.requireNonNull(options);
}
public String getType()
{
if (options.isPreview())
{
switch (options.getPreviewKind())
{
case ALL: return "preview full";
case REPAIRED: return "preview repaired";
case UNREPAIRED: return "preview unrepaired";
case NONE: throw new AssertionError("NONE preview kind not expected when preview repair is set");
default: throw new AssertionError("Unknown preview kind: " + options.getPreviewKind());
}
}
else if (options.accordRepair())
{
return "accord repair";
}
else if (options.isIncremental())
{
return "incremental";
}
return "full";
}
public Collection<SessionState> getSessions()
{
return sessions.values();

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

@ -210,16 +210,17 @@ public class AccordService implements IAccordService, Shutdownable
public void ensureKeyspaceIsAccordManaged(String keyspace) {}
};
private static volatile Node.Id localId = null;
private static volatile IAccordService instance = null;
private static class Handle
@VisibleForTesting
public static void unsafeSetNewAccordService()
{
public static final AccordService instance = new AccordService();
instance = null;
}
public static boolean isSetup()
{
return localId != null;
return instance != null;
}
public static IVerbHandler<? extends Request> verbHandlerOrNoop()
@ -228,22 +229,33 @@ public class AccordService implements IAccordService, Shutdownable
return instance().verbHandler();
}
public static void startup(NodeId tcmId)
public synchronized static void startup(NodeId tcmId)
{
localId = AccordTopologyUtils.tcmIdToAccord(tcmId);
instance().startup();
if (!DatabaseDescriptor.getAccordTransactionsEnabled())
{
instance = NOOP_SERVICE;
return;
}
AccordService as = new AccordService(AccordTopologyUtils.tcmIdToAccord(tcmId));
as.startup();
instance = as;
}
public static void shutdownServiceAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
{
if (localId == null)
IAccordService i = instance;
if (i == null)
return;
instance().shutdownAndWait(timeout, unit);
i.shutdownAndWait(timeout, unit);
}
public static IAccordService instance()
{
return DatabaseDescriptor.getAccordTransactionsEnabled() ? Handle.instance : NOOP_SERVICE;
if (!DatabaseDescriptor.getAccordTransactionsEnabled())
return NOOP_SERVICE;
IAccordService i = instance;
Invariants.checkState(i != null, "AccordService was not started");
return i;
}
public static long uniqueNow()
@ -258,7 +270,7 @@ public class AccordService implements IAccordService, Shutdownable
return timeUnit.convert(Clock.Global.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
private AccordService()
private AccordService(Id localId)
{
Invariants.checkState(localId != null, "static localId must be set before instantiating AccordService");
logger.info("Starting accord with nodeId {}", localId);

13
src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java
View File

@ -19,6 +19,7 @@
package org.apache.cassandra.service.accord;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@ -30,6 +31,7 @@ import java.util.stream.Collectors;
import com.google.common.collect.Sets;
import accord.local.Node;
import accord.primitives.Ranges;
import accord.topology.Shard;
import accord.topology.Topology;
import accord.utils.Invariants;
@ -49,7 +51,7 @@ import org.apache.cassandra.tcm.ownership.DataPlacements;
public class AccordTopologyUtils
{
static Node.Id tcmIdToAccord(NodeId nodeId)
public static Node.Id tcmIdToAccord(NodeId nodeId)
{
return new Node.Id(nodeId.id());
}
@ -95,6 +97,15 @@ public class AccordTopologyUtils
range.right.equals(minToken) ? SentinelKey.max(keyspace) : new TokenKey(keyspace, range.right));
}
public static accord.primitives.Ranges toAccordRanges(String keyspace, Collection<Range<Token>> ranges)
{
List<Range<Token>> normalizedRanges = Range.normalize(ranges);
TokenRange[] tokenRanges = new TokenRange[normalizedRanges.size()];
for (int i = 0; i < normalizedRanges.size(); i++)
tokenRanges[i] = range(keyspace, normalizedRanges.get(i));
return Ranges.of(tokenRanges);
}
public static List<Shard> createShards(KeyspaceMetadata keyspace, DataPlacements placements, Directory directory)
{
ReplicationParams replication = keyspace.params.replication;

10
src/java/org/apache/cassandra/tcm/ClusterMetadata.java
View File

@ -198,6 +198,16 @@ public class ClusterMetadata
this.locator = Locator.usingDirectory(directory);
}
public ClusterMetadata withDirectory(Directory directory)
{
return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordKeyspaces, lockedRanges, inProgressSequences, consensusMigrationState, extensions);
}
public ClusterMetadata withPlacements(DataPlacements placements)
{
return new ClusterMetadata(epoch, partitioner, schema, directory, tokenMap, placements, accordKeyspaces, lockedRanges, inProgressSequences, consensusMigrationState, extensions);
}
public Set<InetAddressAndPort> fullCMSMembers()
{
if (fullCMSEndpoints == null)

37
test/unit/org/apache/cassandra/ServerTestUtils.java
View File

@ -25,8 +25,10 @@ import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -52,6 +54,7 @@ import org.apache.cassandra.locator.BaseProximity;
import org.apache.cassandra.security.ThreadAwareSecurityManager;
import org.apache.cassandra.service.DiskErrorsHandlerService;
import org.apache.cassandra.service.EmbeddedCassandraService;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.tcm.AtomicLongBackedProcessor;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
@ -73,8 +76,10 @@ import org.apache.cassandra.tcm.transformations.UnsafeJoin;
import org.apache.cassandra.tcm.transformations.cms.Initialize;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Sortable;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION;
import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
/**
* Utility methodes used by SchemaLoader and CQLTester to manage the server and its state.
@ -340,6 +345,36 @@ public final class ServerTestUtils
cms.mark();
}
public static void recreateAccord(NodeId tcmid)
{
if (!DatabaseDescriptor.getAccordTransactionsEnabled())
return;
if (AccordService.isSetup())
{
try
{
AccordService.instance().shutdownAndWait(1, TimeUnit.MINUTES);
}
catch (InterruptedException e)
{
throw new UncheckedInterruptedException(e);
}
catch (TimeoutException e)
{
throw new RuntimeException(e);
}
Keyspace ks = Keyspace.open(ACCORD_KEYSPACE_NAME);
FBUtilities.waitOnFutures(ks.flush(ColumnFamilyStore.FlushReason.UNIT_TESTS));
cleanupDirectory(DatabaseDescriptor.getAccordJournalDirectory());
for (ColumnFamilyStore t : ks.getColumnFamilyStores())
t.truncateBlockingWithoutSnapshot();
AccordService.unsafeSetNewAccordService();
}
AccordService.startup(tcmid);
}
public static void markCMS()
{
ClusterMetadataService cms = ClusterMetadataService.instance();

10
test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java
View File

@ -34,6 +34,7 @@ import accord.utilsfork.Gens;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.RetrySpec;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.repair.state.Completable;
import org.apache.cassandra.streaming.StreamEventHandler;
import org.apache.cassandra.streaming.StreamState;
@ -64,7 +65,14 @@ public class FailingRepairFuzzTest extends FuzzTestBase
{
Cluster.Node coordinator = coordinatorGen.next(rs);
RepairCoordinator repair = coordinator.repair(KEYSPACE, repairOption(rs, coordinator, KEYSPACE, TABLES), false);
// exclude accord repair as this test breaks validation/sync; which accord doesn't have
RepairOption options;
do
{
options = repairOption(rs, coordinator, KEYSPACE, TABLES);
}
while (options.accordRepair());
RepairCoordinator repair = coordinator.repair(KEYSPACE, options, false);
repair.run();
InetAddressAndPort failingAddress = pickParticipant(rs, coordinator, repair);
Cluster.Node failingNode = cluster.nodes.get(failingAddress);

137
test/unit/org/apache/cassandra/repair/FuzzTestBase.java
View File

@ -52,7 +52,6 @@ import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.cassandra.config.UnitConfigOverride;
import org.junit.BeforeClass;
import accord.utilsfork.DefaultRandom;
@ -73,6 +72,7 @@ import org.apache.cassandra.concurrent.SimulatedExecutorFactory;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.UnitConfigOverride;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Digest;
@ -99,6 +99,7 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.locator.Locator;
import org.apache.cassandra.locator.RangesAtEndpoint;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.net.ConnectionType;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
@ -116,6 +117,7 @@ import org.apache.cassandra.repair.state.SessionState;
import org.apache.cassandra.repair.state.ValidationState;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.SystemDistributedKeyspace;
import org.apache.cassandra.schema.TableId;
@ -123,6 +125,8 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.Tables;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.AccordConfigurationService;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupComplete;
import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupHistory;
import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupRequest;
@ -137,12 +141,21 @@ import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.streaming.StreamingChannel;
import org.apache.cassandra.streaming.StreamingDataInputPlus;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.listeners.ChangeListener;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.DataPlacements;
import org.apache.cassandra.tools.nodetool.Repair;
import org.apache.cassandra.utils.AbstractTypeGenerators;
import org.apache.cassandra.utils.CassandraGenerators;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.Closeable;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.FailingBiConsumer;
import org.apache.cassandra.utils.Generators;
import org.apache.cassandra.utils.MBeanWrapper;
@ -158,6 +171,7 @@ import org.assertj.core.api.Assertions;
import org.mockito.Mockito;
import org.quicktheories.impl.JavaRandom;
import static org.apache.cassandra.config.CassandraRelevantProperties.ACCORD_REPAIR_RANGE_STEP_UPDATE_INTERVAL;
import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_GLOBAL;
import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION;
@ -177,6 +191,7 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
public static void setUpClass()
{
ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(true);
ACCORD_REPAIR_RANGE_STEP_UPDATE_INTERVAL.setInt(1);
CLOCK_GLOBAL.setString(ClockAccess.class.getName());
// when running in CI an external actor will replace the test configs based off the test type (such as trie, cdc, etc.), this could then have failing tests
// that do not repo with the same seed! To fix that, go to UnitConfigOverride and update the config type to match the one that failed in CI, this should then
@ -288,8 +303,12 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
// so don't want to deal with unlucky histories...
DatabaseDescriptor.setRepairRpcTimeout(TimeUnit.DAYS.toMillis(1));
// make sure accord is enabled as accord has custom repair steps
DatabaseDescriptor.setAccordTransactionsEnabled(true);
InMemory.setUpClass();
MessagingService.instance().listen();
}
public static void setupSchema()
@ -402,15 +421,27 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
for (JobState job : session.getJobs())
{
EnumSet<JobState.State> expected = EnumSet.allOf(JobState.State.class);
if (!shouldSnapshot)
if (repair.state.options.accordRepair())
{
// accord doesn't do snapshot, validation, or streaming
expected.remove(JobState.State.SNAPSHOT_START);
expected.remove(JobState.State.SNAPSHOT_COMPLETE);
}
if (!shouldSync)
{
expected.remove(JobState.State.VALIDATION_START);
expected.remove(JobState.State.VALIDATION_COMPLETE);
expected.remove(JobState.State.STREAM_START);
}
else
{
if (!shouldSnapshot)
{
expected.remove(JobState.State.SNAPSHOT_START);
expected.remove(JobState.State.SNAPSHOT_COMPLETE);
}
if (!shouldSync)
{
expected.remove(JobState.State.STREAM_START);
}
}
Set<JobState.State> actual = job.getStateTimesMillis().keySet();
Assertions.assertThat(actual).isEqualTo(expected);
}
@ -554,11 +585,29 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
private static RepairOption repairOption(RandomSource rs, Cluster.Node coordinator, String ks, Gen<List<String>> tablesGen, Gen<RepairType> repairTypeGen, Gen<PreviewType> previewTypeGen, Gen<RepairParallelism> repairParallelismGen)
{
RepairType type = repairTypeGen.next(rs);
PreviewType previewType = previewTypeGen.next(rs);
boolean accordRepair = type == RepairType.FULL && previewType == PreviewType.NONE ? rs.nextBoolean() : false;
List<String> args = new ArrayList<>();
args.add(ks);
args.addAll(tablesGen.next(rs));
args.add("-pr");
RepairType type = repairTypeGen.next(rs);
List<String> tables = tablesGen.next(rs);
args.addAll(tables);
if (accordRepair)
{
List<Range<Token>> ranges = new ArrayList<>(StorageService.instance.getReplicas(ks, coordinator.broadcastAddressAndPort()).ranges());
ranges.sort(Comparator.naturalOrder());
Range<Token> range = ranges.get(rs.nextInt(0, ranges.size()));
args.add("--start-token");
args.add(range.left.toString());
args.add("--end-token");
Murmur3Partitioner.LongToken left = (Murmur3Partitioner.LongToken) range.left;
Token right = rs.nextBoolean() ? new Murmur3Partitioner.LongToken(left.token + 100) : range.right;
args.add(right.toString());
}
else
{
args.add("-pr");
}
switch (type)
{
case IR:
@ -570,7 +619,6 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
default:
throw new AssertionError("Unsupported repair type: " + type);
}
PreviewType previewType = previewTypeGen.next(rs);
switch (previewType)
{
case NONE:
@ -601,6 +649,8 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
}
if (rs.nextBoolean()) args.add("--optimise-streams");
RepairOption options = RepairOption.parse(Repair.parseOptionMap(() -> "test", args), DatabaseDescriptor.getPartitioner());
if (accordRepair)
options = options.withAccordRepair(true);
if (options.getRanges().isEmpty())
{
if (options.isPrimaryRange())
@ -757,9 +807,71 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
ClusterMetadataTestHelper.register(inst.broadcastAddressAndPort());
ClusterMetadataTestHelper.join(inst.broadcastAddressAndPort(), inst.tokens());
}
List<InetAddressAndPort> addresses = new ArrayList<>(nodes.keySet());
addresses.sort(Comparator.naturalOrder());
NodeId tcmid = ClusterMetadata.current().directory.peerId(addresses.get(rs.nextInt(0, addresses.size())));
ServerTestUtils.recreateAccord(tcmid);
interceptTCMNotifications(tcmid);
setupSchema();
}
private void interceptTCMNotifications(NodeId tcmid)
{
AccordService as = (AccordService) AccordService.instance();
AccordConfigurationService config = as.configurationService();
ClusterMetadataService.instance().log().removeListener(config);
ClusterMetadataService.instance().log().addListener(new ChangeListener()
{
@Override
public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
{
config.notifyPostCommit(sanitize(prev, tcmid), sanitize(next, tcmid), fromSnapshot);
}
});
}
private ClusterMetadata sanitize(ClusterMetadata metadata, NodeId tcmid)
{
if (metadata.directory.isEmpty())
return metadata;
ClusterMetadata sanitized = metadata.withDirectory(sanitize(metadata.directory, tcmid))
.withPlacements(sanitize(metadata.placements, FBUtilities.getBroadcastAddressAndPort()));
return sanitized;
}
private Directory sanitize(Directory directory, NodeId tcmid)
{
if (directory.getNodeAddresses(tcmid) == null)
throw new AssertionError("Expected node " + tcmid + " but not found in " + directory);
for (NodeId peer : directory.peerIds())
{
if (peer.equals(tcmid))
continue;
directory = directory.without(peer);
}
directory = directory.withNodeAddresses(tcmid, NodeAddresses.current());
return directory;
}
private DataPlacements sanitize(DataPlacements placements, InetAddressAndPort endpoint)
{
DataPlacements.Builder builder = DataPlacements.builder(placements.size());
for (Map.Entry<ReplicationParams, DataPlacement> e : placements)
builder.with(e.getKey(), sanitize(placements.lastModified(), e.getValue(), endpoint));
return builder.build();
}
private DataPlacement sanitize(Epoch epoch, DataPlacement value, InetAddressAndPort endpoint)
{
DataPlacement.Builder builder = DataPlacement.builder();
for (Range<Token> e : value.writes.ranges())
builder.withWriteReplica(epoch, new Replica(endpoint, e, true));
for (Range<Token> e : value.reads.ranges())
builder.withReadReplica(epoch, new Replica(endpoint, e, true));
return builder.build();
}
public Closeable addListener(MessageListener listener)
{
listeners.add(listener);
@ -1295,6 +1407,8 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
failures.add(new AssertionError(event.getMessage()));
});
}
if (repair.state.options.accordRepair())
AccordService.instance().ensureKeyspaceIsAccordManaged(repair.state.keyspace);
return repair;
}
@ -1449,7 +1563,10 @@ public abstract class FuzzTestBase extends CQLTester.InMemory
if (("org.apache.cassandra.service.paxos.Paxos".equals(next.getClassName()) && "newBallot".equals(next.getMethodName()))
|| ("org.apache.cassandra.service.paxos.uncommitted.PaxosBallotTracker".equals(next.getClassName()) && "updateLowBound".equals(next.getMethodName())))
return Access.MAIN_THREAD_ONLY;
if (next.getClassName().startsWith("org.apache.cassandra.db.")
if (next.getClassName().startsWith("org.apache.cassandra.accord.")
|| next.getClassName().startsWith("org.apache.cassandra.journal.")
|| next.getClassName().startsWith("org.apache.cassandra.service.accord.")
|| next.getClassName().startsWith("org.apache.cassandra.db.")
|| next.getClassName().startsWith("org.apache.cassandra.gms.")
|| next.getClassName().startsWith("org.apache.cassandra.cql3.")
|| next.getClassName().startsWith("org.apache.cassandra.metrics.")

29
test/unit/org/apache/cassandra/repair/HappyPathFuzzTest.java
View File

@ -19,12 +19,18 @@
package org.apache.cassandra.repair;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.LongStream;
import org.junit.Test;
import accord.utilsfork.Gen;
import accord.utilsfork.Gens;
import org.agrona.collections.LongArrayList;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.RetrySpec;
import org.apache.cassandra.utils.Closeable;
@ -38,6 +44,8 @@ public class HappyPathFuzzTest extends FuzzTestBase
{
// disable all retries, no delays/drops are possible
DatabaseDescriptor.getRepairRetrySpec().maxAttempts = RetrySpec.MaxAttempt.DISABLED;
Map<String, LongArrayList> repairTypeRuntimes = new HashMap<>();
long realStartNanos = System.nanoTime();
qt().withPure(false).withExamples(10).check(rs -> {
Cluster cluster = new Cluster(rs);
Gen<Cluster.Node> coordinatorGen = Gens.pick(cluster.nodes.keySet()).map(cluster.nodes::get);
@ -47,6 +55,7 @@ public class HappyPathFuzzTest extends FuzzTestBase
{
Cluster.Node coordinator = coordinatorGen.next(rs);
long nowNanos = System.nanoTime();
RepairCoordinator repair = coordinator.repair(KEYSPACE, repairOption(rs, coordinator, KEYSPACE, TABLES));
repair.run();
boolean shouldSync = rs.nextBoolean();
@ -54,9 +63,29 @@ public class HappyPathFuzzTest extends FuzzTestBase
closeables.add(cluster.nodes.get(pickParticipant(rs, coordinator, repair)).doValidation((cfs, validator) -> addMismatch(rs, cfs, validator)));
runAndAssertSuccess(cluster, example, shouldSync, repair);
repairTypeRuntimes.computeIfAbsent(repair.state.getType(), ignore -> new LongArrayList()).addLong(System.nanoTime() - nowNanos);
closeables.forEach(Closeable::close);
closeables.clear();
}
});
long realDurationNanos = System.nanoTime() - realStartNanos;
long repairDurationsNanos = 0;
StringBuilder sb = new StringBuilder();
for (Map.Entry<String, LongArrayList> e : repairTypeRuntimes.entrySet())
{
sb.append(e.getKey());
long[] times = e.getValue().toLongArray();
repairDurationsNanos += LongStream.of(times).sum();
Arrays.sort(times);
long min = times[0];
long median = times[times.length / 2];
long max = times[times.length - 1];
sb.append(": min=").append(TimeUnit.NANOSECONDS.toMillis(min))
.append(", median=").append(TimeUnit.NANOSECONDS.toMillis(median))
.append(", max=").append(TimeUnit.NANOSECONDS.toMillis(max))
.append(", count=").append(times.length)
.append('\n');
}
logger.info("Repair runtimes (in millis):\nTest Duration {}\nRepair Duration {}\n{}", TimeUnit.NANOSECONDS.toMillis(realDurationNanos), TimeUnit.NANOSECONDS.toMillis(repairDurationsNanos), sb);
}
}