From a9cb71833a0009e09a40dd869673d3205fc50d89 Mon Sep 17 00:00:00 2001 From: Ariel Weisberg Date: Wed, 29 Nov 2023 17:12:46 -0500 Subject: [PATCH] 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 --- modules/accord | 2 +- .../apache/cassandra/config/AccordSpec.java | 15 ++ .../config/CassandraRelevantProperties.java | 1 + .../org/apache/cassandra/config/Config.java | 15 -- .../cassandra/config/DatabaseDescriptor.java | 8 +- .../db/virtual/LocalRepairTables.java | 26 +--- .../cassandra/repair/AbstractRepairJob.java | 2 +- .../cassandra/repair/AccordRepairJob.java | 43 +++--- .../cassandra/repair/RepairCoordinator.java | 1 + .../repair/messages/RepairOption.java | 10 ++ .../repair/state/CoordinatorState.java | 24 +++ .../service/accord/AccordService.java | 34 +++-- .../service/accord/AccordTopologyUtils.java | 13 +- .../apache/cassandra/tcm/ClusterMetadata.java | 10 ++ .../org/apache/cassandra/ServerTestUtils.java | 37 ++++- .../repair/FailingRepairFuzzTest.java | 10 +- .../apache/cassandra/repair/FuzzTestBase.java | 137 ++++++++++++++++-- .../cassandra/repair/HappyPathFuzzTest.java | 29 ++++ 18 files changed, 330 insertions(+), 87 deletions(-) diff --git a/modules/accord b/modules/accord index d9ef555302..3ca9e55024 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit d9ef555302f8774ed03325ba22d38ee0b80130a8 +Subproject commit 3ca9e55024192e9b7c38ad5330229830343b74f2 diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java index b025d561b6..2143342bd4 100644 --- a/src/java/org/apache/cassandra/config/AccordSpec.java +++ b/src/java/org/apache/cassandra/config/AccordSpec.java @@ -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"); } diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index c79e7e2746..f1f50e589f 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -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"), diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 2dd95248fe..3352ce9fed 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -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"); diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 757407ad7b..12da2dae42 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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) diff --git a/src/java/org/apache/cassandra/db/virtual/LocalRepairTables.java b/src/java/org/apache/cassandra/db/virtual/LocalRepairTables.java index 0012382bd9..34aeccf128 100644 --- a/src/java/org/apache/cassandra/db/virtual/LocalRepairTables.java +++ b/src/java/org/apache/cassandra/db/virtual/LocalRepairTables.java @@ -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 diff --git a/src/java/org/apache/cassandra/repair/AbstractRepairJob.java b/src/java/org/apache/cassandra/repair/AbstractRepairJob.java index 7c4346e3bf..df3a67dbc9 100644 --- a/src/java/org/apache/cassandra/repair/AbstractRepairJob.java +++ b/src/java/org/apache/cassandra/repair/AbstractRepairJob.java @@ -28,7 +28,7 @@ import org.apache.cassandra.utils.concurrent.AsyncFuture; public abstract class AbstractRepairJob extends AsyncFuture implements Runnable { - private final SharedContext ctx; + protected final SharedContext ctx; public final JobState state; protected final RepairJobDesc desc; protected final RepairSession session; diff --git a/src/java/org/apache/cassandra/repair/AccordRepairJob.java b/src/java/org/apache/cassandra/repair/AccordRepairJob.java index 1736d799fa..c2e9eb792e 100644 --- a/src/java/org/apache/cassandra/repair/AccordRepairJob.java +++ b/src/java/org/apache/cassandra/repair/AccordRepairJob.java @@ -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> 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()); diff --git a/src/java/org/apache/cassandra/repair/RepairCoordinator.java b/src/java/org/apache/cassandra/repair/RepairCoordinator.java index 79159c9fe1..c2d2415a7c 100644 --- a/src/java/org/apache/cassandra/repair/RepairCoordinator.java +++ b/src/java/org/apache/cassandra/repair/RepairCoordinator.java @@ -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 diff --git a/src/java/org/apache/cassandra/repair/messages/RepairOption.java b/src/java/org/apache/cassandra/repair/messages/RepairOption.java index bef7acfe16..626c2d18b5 100644 --- a/src/java/org/apache/cassandra/repair/messages/RepairOption.java +++ b/src/java/org/apache/cassandra/repair/messages/RepairOption.java @@ -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; diff --git a/src/java/org/apache/cassandra/repair/state/CoordinatorState.java b/src/java/org/apache/cassandra/repair/state/CoordinatorState.java index 5bc8a9e5d8..43d17acca6 100644 --- a/src/java/org/apache/cassandra/repair/state/CoordinatorState.java +++ b/src/java/org/apache/cassandra/repair/state/CoordinatorState.java @@ -64,6 +64,30 @@ public class CoordinatorState extends AbstractState getSessions() { return sessions.values(); diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index fd115f8943..ce0f6f6f1e 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -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 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); diff --git a/src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java b/src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java index d8b757941a..e88587411b 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTopologyUtils.java @@ -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> ranges) + { + List> 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 createShards(KeyspaceMetadata keyspace, DataPlacements placements, Directory directory) { ReplicationParams replication = keyspace.params.replication; diff --git a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java index 2b9dd0a04d..8dfe9a6790 100644 --- a/src/java/org/apache/cassandra/tcm/ClusterMetadata.java +++ b/src/java/org/apache/cassandra/tcm/ClusterMetadata.java @@ -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 fullCMSMembers() { if (fullCMSEndpoints == null) diff --git a/test/unit/org/apache/cassandra/ServerTestUtils.java b/test/unit/org/apache/cassandra/ServerTestUtils.java index 130053658e..3379743452 100644 --- a/test/unit/org/apache/cassandra/ServerTestUtils.java +++ b/test/unit/org/apache/cassandra/ServerTestUtils.java @@ -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(); diff --git a/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java b/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java index cc0d781898..6ee81d1545 100644 --- a/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java +++ b/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/repair/FuzzTestBase.java b/test/unit/org/apache/cassandra/repair/FuzzTestBase.java index 9eafb67d9b..7dcf4c6404 100644 --- a/test/unit/org/apache/cassandra/repair/FuzzTestBase.java +++ b/test/unit/org/apache/cassandra/repair/FuzzTestBase.java @@ -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 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 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> tablesGen, Gen repairTypeGen, Gen previewTypeGen, Gen repairParallelismGen) { + RepairType type = repairTypeGen.next(rs); + PreviewType previewType = previewTypeGen.next(rs); + boolean accordRepair = type == RepairType.FULL && previewType == PreviewType.NONE ? rs.nextBoolean() : false; List args = new ArrayList<>(); args.add(ks); - args.addAll(tablesGen.next(rs)); - args.add("-pr"); - RepairType type = repairTypeGen.next(rs); + List tables = tablesGen.next(rs); + args.addAll(tables); + if (accordRepair) + { + List> ranges = new ArrayList<>(StorageService.instance.getReplicas(ks, coordinator.broadcastAddressAndPort()).ranges()); + ranges.sort(Comparator.naturalOrder()); + Range 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 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 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 e : value.writes.ranges()) + builder.withWriteReplica(epoch, new Replica(endpoint, e, true)); + for (Range 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.") diff --git a/test/unit/org/apache/cassandra/repair/HappyPathFuzzTest.java b/test/unit/org/apache/cassandra/repair/HappyPathFuzzTest.java index b145e28fb9..9809421931 100644 --- a/test/unit/org/apache/cassandra/repair/HappyPathFuzzTest.java +++ b/test/unit/org/apache/cassandra/repair/HappyPathFuzzTest.java @@ -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 repairTypeRuntimes = new HashMap<>(); + long realStartNanos = System.nanoTime(); qt().withPure(false).withExamples(10).check(rs -> { Cluster cluster = new Cluster(rs); Gen 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 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); } }