diff --git a/CHANGES.txt b/CHANGES.txt index 4b8393ebc2..44ca779885 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -8,6 +8,7 @@ * Allow empty keystore_password in encryption_options (CASSANDRA-18778) * Skip ColumnFamilyStore#topPartitions initialization when client or tool mode (CASSANDRA-18697) Merged from 4.0: + * Add new concurrent_merkle_tree_requests config property to prevent OOM during multi-range and/or multi-table repairs (CASSANDRA-19336) * Skip version check if an endpoint is dead state in Gossiper#upgradeFromVersionSupplier (CASSANDRA-19187) * Fix Gossiper::hasMajorVersion3Nodes to return false during minor upgrade (CASSANDRA-18999) * Revert unnecessary read lock acquisition when reading ring version in TokenMetadata introduced in CASSANDRA-16286 (CASSANDRA-19107) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 4b2711cfb7..1986d6fa29 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -643,18 +643,42 @@ concurrent_materialized_view_writes: 32 # off heap objects memtable_allocation_type: heap_buffers -# Limit memory usage for Merkle tree calculations during repairs. The default -# is 1/16th of the available heap. The main tradeoff is that smaller trees -# have less resolution, which can lead to over-streaming data. If you see heap -# pressure during repairs, consider lowering this, but you cannot go below -# one mebibyte. If you see lots of over-streaming, consider raising -# this or using subrange repair. +# Limit memory usage for Merkle tree calculations during repairs of a certain +# table and common token range. Repair commands targetting multiple tables or +# virtual nodes can exceed this limit depending on concurrent_merkle_tree_requests. +# +# The default is 1/16th of the available heap. The main tradeoff is that +# smaller trees have less resolution, which can lead to over-streaming data. +# If you see heap pressure during repairs, consider lowering this, but you +# cannot go below one mebibyte. If you see lots of over-streaming, consider +# raising this or using subrange repair. # # For more details see https://issues.apache.org/jira/browse/CASSANDRA-14096. # # Min unit: MiB # repair_session_space: +# The number of simultaneous Merkle tree requests during repairs that can +# be performed by a repair command. The size of each validation request is +# limited by the repair_session_space property, so setting this to 1 will make +# sure that a repair command doesn't exceed that limit, even if the repair +# command is repairing multiple tables or multiple virtual nodes. +# +# There isn't a limit by default for backwards compatibility, but this can +# produce OOM for commands repairing multiple tables or multiple virtual nodes. +# A limit of just 1 simultaneous Merkle tree request is generally recommended +# with no virtual nodes so repair_session_space, and thereof the Merkle tree +# resolution, can be high. For virtual nodes a value of 1 with the default +# repair_session_space value will produce higher resolution Merkle trees +# at the expense of speed. Alternatively, when working with virtual nodes it +# can make sense to reduce the repair_session_space and increase the value of +# concurrent_merkle_tree_requests because each range will contain fewer data. +# +# For more details see https://issues.apache.org/jira/browse/CASSANDRA-19336. +# +# A zero value means no limit. +# concurrent_merkle_tree_requests: 0 + # Total space to use for commit logs on disk. # # If space gets above this value, Cassandra will flush every dirty CF diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 8a59ca2cda..298903acf3 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -195,6 +195,8 @@ public class Config @Replaces(oldName = "repair_session_space_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_INT, deprecated = true) public volatile DataStorageSpec.IntMebibytesBound repair_session_space = null; + public volatile int concurrent_merkle_tree_requests = 0; + public volatile boolean use_offheap_merkle_trees = true; public int storage_port = 7000; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index a04e85c1bc..0805b1f9ec 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -3417,6 +3417,16 @@ public class DatabaseDescriptor conf.repair_session_space = new DataStorageSpec.IntMebibytesBound(sizeInMiB); } + public static int getConcurrentMerkleTreeRequests() + { + return conf.concurrent_merkle_tree_requests; + } + + public static void setConcurrentMerkleTreeRequests(int value) + { + conf.concurrent_merkle_tree_requests = value; + } + public static int getPaxosRepairParallelism() { return conf.paxos_repair_parallelism; diff --git a/src/java/org/apache/cassandra/repair/AbstractRepairTask.java b/src/java/org/apache/cassandra/repair/AbstractRepairTask.java index d2a6f1a786..f2449f703a 100644 --- a/src/java/org/apache/cassandra/repair/AbstractRepairTask.java +++ b/src/java/org/apache/cassandra/repair/AbstractRepairTask.java @@ -55,6 +55,7 @@ public abstract class AbstractRepairTask implements RepairTask private List submitRepairSessions(TimeUUID parentSession, boolean isIncremental, ExecutorPlus executor, + Scheduler validationScheduler, List commonRanges, String... cfnames) { @@ -74,7 +75,9 @@ public abstract class AbstractRepairTask implements RepairTask options.repairPaxos(), options.paxosOnly(), executor, + validationScheduler, cfnames); + if (session == null) continue; session.addCallback(new RepairSessionCallback(session)); @@ -86,10 +89,11 @@ public abstract class AbstractRepairTask implements RepairTask protected Future runRepair(TimeUUID parentSession, boolean isIncremental, ExecutorPlus executor, + Scheduler validationScheduler, List commonRanges, String... cfnames) { - List allSessions = submitRepairSessions(parentSession, isIncremental, executor, commonRanges, cfnames); + List allSessions = submitRepairSessions(parentSession, isIncremental, executor, validationScheduler, commonRanges, cfnames); List>> ranges = Lists.transform(allSessions, RepairSession::ranges); Future> f = FutureCombiner.successfulOf(allSessions); return f.map(results -> { diff --git a/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java b/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java index af1a234d34..6cfd6ff84b 100644 --- a/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java +++ b/src/java/org/apache/cassandra/repair/IncrementalRepairTask.java @@ -57,7 +57,7 @@ public class IncrementalRepairTask extends AbstractRepairTask } @Override - public Future performUnsafe(ExecutorPlus executor) throws Exception + public Future performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) throws Exception { // the local node also needs to be included in the set of participants, since coordinator sessions aren't persisted Set allParticipants = ImmutableSet.builder() @@ -69,7 +69,7 @@ public class IncrementalRepairTask extends AbstractRepairTask CoordinatorSession coordinatorSession = ActiveRepairService.instance.consistent.coordinated.registerSession(parentSession, allParticipants, neighborsAndRanges.shouldExcludeDeadParticipants); - return coordinatorSession.execute(() -> runRepair(parentSession, true, executor, allRanges, cfnames)); + return coordinatorSession.execute(() -> runRepair(parentSession, true, executor, validationScheduler, allRanges, cfnames)); } } diff --git a/src/java/org/apache/cassandra/repair/NormalRepairTask.java b/src/java/org/apache/cassandra/repair/NormalRepairTask.java index 56a03f8816..a42b38012e 100644 --- a/src/java/org/apache/cassandra/repair/NormalRepairTask.java +++ b/src/java/org/apache/cassandra/repair/NormalRepairTask.java @@ -50,8 +50,8 @@ public class NormalRepairTask extends AbstractRepairTask } @Override - public Future performUnsafe(ExecutorPlus executor) + public Future performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) { - return runRepair(parentSession, false, executor, commonRanges, cfnames); + return runRepair(parentSession, false, executor, validationScheduler, commonRanges, cfnames); } } diff --git a/src/java/org/apache/cassandra/repair/PreviewRepairTask.java b/src/java/org/apache/cassandra/repair/PreviewRepairTask.java index 7ce7d1fbba..c97632ef8f 100644 --- a/src/java/org/apache/cassandra/repair/PreviewRepairTask.java +++ b/src/java/org/apache/cassandra/repair/PreviewRepairTask.java @@ -64,9 +64,9 @@ public class PreviewRepairTask extends AbstractRepairTask } @Override - public Future performUnsafe(ExecutorPlus executor) + public Future performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) { - Future f = runRepair(parentSession, false, executor, commonRanges, cfnames); + Future f = runRepair(parentSession, false, executor, validationScheduler, commonRanges, cfnames); return f.map(result -> { if (result.hasFailed()) return result; diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java b/src/java/org/apache/cassandra/repair/RepairJob.java index aba8bd8c85..d6f8820e56 100644 --- a/src/java/org/apache/cassandra/repair/RepairJob.java +++ b/src/java/org/apache/cassandra/repair/RepairJob.java @@ -110,7 +110,7 @@ public class RepairJob extends AsyncFuture implements Runnable /** * Runs repair job. - * + *

* This sets up necessary task and runs them on given {@code taskExecutor}. * After submitting all tasks, waits until validation with replica completes. */ @@ -123,7 +123,6 @@ public class RepairJob extends AsyncFuture implements Runnable List allEndpoints = new ArrayList<>(session.state.commonRange.endpoints); allEndpoints.add(FBUtilities.getBroadcastAddressAndPort()); - Future> treeResponses; Future paxosRepair; if (paxosRepairEnabled() && ((useV2() && session.repairPaxos) || session.paxosOnly)) { @@ -160,9 +159,9 @@ public class RepairJob extends AsyncFuture implements Runnable } // Create a snapshot at all nodes unless we're using pure parallel repairs + final Future allSnapshotTasks; if (parallelismDegree != RepairParallelism.PARALLEL) { - Future allSnapshotTasks; if (session.isIncremental) { // consistent repair does it's own "snapshotting" @@ -186,27 +185,16 @@ public class RepairJob extends AsyncFuture implements Runnable }); }); } - - // When all snapshot complete, send validation requests - treeResponses = allSnapshotTasks.flatMap(endpoints -> { - if (parallelismDegree == RepairParallelism.SEQUENTIAL) - return sendSequentialValidationRequest(allEndpoints); - else - return sendDCAwareValidationRequest(allEndpoints); - }, taskExecutor); } else { - // If not sequential, just send validation request to all replica - treeResponses = paxosRepair.flatMap(input -> sendValidationRequest(allEndpoints)); + allSnapshotTasks = null; } - treeResponses = treeResponses.map(a -> { - state.phase.validationCompleted(); - return a; - }); - // When all validations complete, submit sync tasks - Future> syncResults = treeResponses.flatMap(session.optimiseStreams && !session.pullRepair ? this::optimisedSyncing : this::standardSyncing, taskExecutor); + // Run validations and the creation of sync tasks in the scheduler, so it can limit the number of Merkle trees + // that there are in memory at once. When all validations complete, submit sync tasks out of the scheduler. + Future> syncResults = session.validationScheduler.schedule(() -> createSyncTasks(paxosRepair, allSnapshotTasks, allEndpoints), taskExecutor) + .flatMap(this::executeTasks, taskExecutor); // When all sync complete, set the final result syncResults.addCallback(new FutureCallback>() @@ -248,24 +236,52 @@ public class RepairJob extends AsyncFuture implements Runnable }, taskExecutor); } + private Future> createSyncTasks(Future paxosRepair, Future allSnapshotTasks, List allEndpoints) + { + Future> treeResponses; + if (allSnapshotTasks != null) + { + // When all snapshot complete, send validation requests + treeResponses = allSnapshotTasks.flatMap(endpoints -> { + if (parallelismDegree == RepairParallelism.SEQUENTIAL) + return sendSequentialValidationRequest(allEndpoints); + else + return sendDCAwareValidationRequest(allEndpoints); + }, taskExecutor); + } + else + { + // If not sequential, just send validation request to all replica + treeResponses = paxosRepair.flatMap(input -> sendValidationRequest(allEndpoints)); + } + + treeResponses = treeResponses.map(a -> { + state.phase.validationCompleted(); + return a; + }); + + return treeResponses.map(session.optimiseStreams && !session.pullRepair + ? this::createOptimisedSyncingSyncTasks + : this::createStandardSyncTasks, taskExecutor); + } + private boolean isTransient(InetAddressAndPort ep) { return session.state.commonRange.transEndpoints.contains(ep); } - private Future> standardSyncing(List trees) + private List createStandardSyncTasks(List trees) { - state.phase.streamSubmitted(); - List syncTasks = createStandardSyncTasks(desc, - trees, - FBUtilities.getLocalAddressAndPort(), - this::isTransient, - session.isIncremental, - session.pullRepair, - session.previewKind); - return executeTasks(syncTasks); + return createStandardSyncTasks(desc, + trees, + FBUtilities.getLocalAddressAndPort(), + this::isTransient, + session.isIncremental, + session.pullRepair, + session.previewKind); } + @VisibleForTesting static List createStandardSyncTasks(RepairJobDesc desc, List trees, InetAddressAndPort local, @@ -333,20 +349,6 @@ public class RepairJob extends AsyncFuture implements Runnable return syncTasks; } - private Future> optimisedSyncing(List trees) - { - state.phase.streamSubmitted(); - List syncTasks = createOptimisedSyncingSyncTasks(desc, - trees, - FBUtilities.getLocalAddressAndPort(), - this::isTransient, - this::getDC, - session.isIncremental, - session.previewKind); - - return executeTasks(syncTasks); - } - @VisibleForTesting Future> executeTasks(List tasks) { @@ -385,6 +387,17 @@ public class RepairJob extends AsyncFuture implements Runnable } } + private List createOptimisedSyncingSyncTasks(List trees) + { + return createOptimisedSyncingSyncTasks(desc, + trees, + FBUtilities.getLocalAddressAndPort(), + this::isTransient, + this::getDC, + session.isIncremental, + session.previewKind); + } + static List createOptimisedSyncingSyncTasks(RepairJobDesc desc, List trees, InetAddressAndPort local, @@ -537,12 +550,7 @@ public class RepairJob extends AsyncFuture implements Runnable for (InetAddressAndPort endpoint : endpoints) { String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint); - Queue queue = requestsByDatacenter.get(dc); - if (queue == null) - { - queue = new LinkedList<>(); - requestsByDatacenter.put(dc, queue); - } + Queue queue = requestsByDatacenter.computeIfAbsent(dc, k -> new LinkedList<>()); queue.add(endpoint); } diff --git a/src/java/org/apache/cassandra/repair/RepairRunnable.java b/src/java/org/apache/cassandra/repair/RepairRunnable.java index 7607b045fa..a56a32a53d 100644 --- a/src/java/org/apache/cassandra/repair/RepairRunnable.java +++ b/src/java/org/apache/cassandra/repair/RepairRunnable.java @@ -102,6 +102,7 @@ public class RepairRunnable implements Runnable, ProgressEventNotifier, RepairNo private final List listeners = new ArrayList<>(); private final AtomicReference firstError = new AtomicReference<>(null); + final Scheduler validationScheduler; private TraceState traceState; @@ -109,6 +110,7 @@ public class RepairRunnable implements Runnable, ProgressEventNotifier, RepairNo { this.state = new CoordinatorState(cmd, keyspace, options); this.storageService = storageService; + this.validationScheduler = Scheduler.build(DatabaseDescriptor.getConcurrentMerkleTreeRequests()); this.tag = "repair:" + cmd; ActiveRepairService.instance.register(state); @@ -420,7 +422,7 @@ public class RepairRunnable implements Runnable, ProgressEventNotifier, RepairNo ExecutorPlus executor = createExecutor(); state.phase.repairSubmitted(); - Future f = task.perform(executor); + Future f = task.perform(executor, validationScheduler); f.addCallback((result, failure) -> { state.phase.repairCompleted(); try diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java index 6fb455b47a..5e86a3922b 100644 --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@ -127,6 +127,7 @@ public class RepairSession extends AsyncFuture implements I // Tasks(snapshot, validate request, differencing, ...) are run on taskExecutor public final ExecutorPlus taskExecutor; public final boolean optimiseStreams; + public final Scheduler validationScheduler; private volatile boolean terminated = false; @@ -142,6 +143,7 @@ public class RepairSession extends AsyncFuture implements I * @param cfnames names of columnfamilies */ public RepairSession(TimeUUID parentRepairSession, + Scheduler validationScheduler, CommonRange commonRange, String keyspace, RepairParallelism parallelismDegree, @@ -153,6 +155,7 @@ public class RepairSession extends AsyncFuture implements I boolean paxosOnly, String... cfnames) { + this.validationScheduler = validationScheduler; this.repairPaxos = repairPaxos; this.paxosOnly = paxosOnly; assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it"; diff --git a/src/java/org/apache/cassandra/repair/RepairTask.java b/src/java/org/apache/cassandra/repair/RepairTask.java index dc71d6e1b5..8e3c0bfb64 100644 --- a/src/java/org/apache/cassandra/repair/RepairTask.java +++ b/src/java/org/apache/cassandra/repair/RepairTask.java @@ -31,13 +31,13 @@ public interface RepairTask return name() + " completed successfully"; } - Future performUnsafe(ExecutorPlus executor) throws Exception; + Future performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) throws Exception; - default Future perform(ExecutorPlus executor) + default Future perform(ExecutorPlus executor, Scheduler validationScheduler) { try { - return performUnsafe(executor); + return performUnsafe(executor, validationScheduler); } catch (Exception | Error e) { diff --git a/src/java/org/apache/cassandra/repair/Scheduler.java b/src/java/org/apache/cassandra/repair/Scheduler.java new file mode 100644 index 0000000000..cc08a0666c --- /dev/null +++ b/src/java/org/apache/cassandra/repair/Scheduler.java @@ -0,0 +1,118 @@ +/* + * 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.repair; + +import java.util.LinkedList; +import java.util.Queue; +import java.util.concurrent.Executor; +import java.util.function.Supplier; +import javax.annotation.concurrent.GuardedBy; + +import org.apache.cassandra.utils.Pair; +import org.apache.cassandra.utils.concurrent.AsyncFuture; +import org.apache.cassandra.utils.concurrent.Future; + +/** + * Task scheduler that limits the number of concurrent tasks across multiple executors. + */ +public interface Scheduler +{ + default Future schedule(Supplier> task, Executor executor) + { + return schedule(new Task<>(task), executor); + } + + Task schedule(Task task, Executor executor); + + static Scheduler build(int concurrentValidations) + { + return concurrentValidations <= 0 + ? new NoopScheduler() + : new LimitedConcurrentScheduler(concurrentValidations); + } + + final class NoopScheduler implements Scheduler + { + @Override + public Task schedule(Task task, Executor executor) + { + executor.execute(task); + return task; + } + } + + final class LimitedConcurrentScheduler implements Scheduler + { + private final int concurrentValidations; + @GuardedBy("this") + private int inflight = 0; + @GuardedBy("this") + private final Queue, Executor>> tasks = new LinkedList<>(); + + LimitedConcurrentScheduler(int concurrentValidations) + { + this.concurrentValidations = concurrentValidations; + } + + @Override + public synchronized Task schedule(Task task, Executor executor) + { + tasks.offer(Pair.create(task, executor)); + maybeSchedule(); + return task; + } + + private synchronized void onDone() + { + inflight--; + maybeSchedule(); + } + + private void maybeSchedule() + { + if (inflight == concurrentValidations || tasks.isEmpty()) + return; + inflight++; + Pair, Executor> pair = tasks.poll(); + pair.left.addCallback((s, f) -> onDone()); + pair.right.execute(pair.left); + } + } + + class Task extends AsyncFuture implements Runnable + { + private final Supplier> supplier; + + public Task(Supplier> supplier) + { + this.supplier = supplier; + } + + @Override + public void run() + { + supplier.get().addCallback((s, f) -> { + if (f != null) + tryFailure(f); + else + trySuccess(s); + }); + } + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java index a0716b17df..eed52b78f8 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@ -45,6 +45,7 @@ import org.apache.cassandra.concurrent.ExecutorPlus; import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DurationSpec; import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.repair.Scheduler; import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.EndpointsByRange; import org.apache.cassandra.locator.EndpointsForRange; @@ -306,6 +307,19 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai return DatabaseDescriptor.getRepairSessionSpaceInMiB(); } + @Override + public int getConcurrentMerkleTreeRequests() + { + return DatabaseDescriptor.getConcurrentMerkleTreeRequests(); + } + + @Override + public void setConcurrentMerkleTreeRequests(int value) + { + logger.info("Setting concurrent_merkle_tree_requests to {}", value); + DatabaseDescriptor.setConcurrentMerkleTreeRequests(value); + } + public List getRepairStats(List schemaArgs, String rangeString) { List stats = new ArrayList<>(); @@ -387,6 +401,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai boolean repairPaxos, boolean paxosOnly, ExecutorPlus executor, + Scheduler validationScheduler, String... cfnames) { if (repairPaxos && previewKind != PreviewKind.NONE) @@ -398,7 +413,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai if (cfnames.length == 0) return null; - final RepairSession session = new RepairSession(parentRepairSession, range, keyspace, + final RepairSession session = new RepairSession(parentRepairSession, validationScheduler, range, keyspace, parallelismDegree, isIncremental, pullRepair, previewKind, optimiseStreams, repairPaxos, paxosOnly, cfnames); repairs.getIfPresent(parentRepairSession).register(session.state); diff --git a/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java b/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java index 009ad562af..532b17c6f5 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java @@ -37,6 +37,10 @@ public interface ActiveRepairServiceMBean public void setRepairSessionSpaceInMebibytes(int sizeInMebibytes); public int getRepairSessionSpaceInMebibytes(); + int getConcurrentMerkleTreeRequests(); + + void setConcurrentMerkleTreeRequests(int value); + public boolean getUseOffheapMerkleTrees(); public void setUseOffheapMerkleTrees(boolean value); diff --git a/test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java index 14d49dc0ca..e2529b4915 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java @@ -29,8 +29,6 @@ import java.util.Map; import java.util.Random; import java.util.Set; import java.util.concurrent.Callable; -import java.util.function.Function; -import java.util.function.Predicate; import java.util.concurrent.TimeoutException; import org.junit.Test; @@ -48,12 +46,11 @@ import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.repair.AsymmetricRemoteSyncTask; import org.apache.cassandra.repair.LocalSyncTask; import org.apache.cassandra.repair.RepairJob; -import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.SyncTask; import org.apache.cassandra.repair.TreeResponse; -import org.apache.cassandra.streaming.PreviewKind; import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NETWORK; import static org.junit.Assert.assertEquals; @@ -108,22 +105,17 @@ public class OptimiseStreamsRepairTest extends TestBaseImpl public static void install(ClassLoader cl, int id) { new ByteBuddy().rebase(RepairJob.class) - .method(named("createOptimisedSyncingSyncTasks")) + .method(named("createOptimisedSyncingSyncTasks").and(takesArguments(1))) .intercept(MethodDelegation.to(BBHelper.class)) .make() .load(cl, ClassLoadingStrategy.Default.INJECTION); } - public static List createOptimisedSyncingSyncTasks(RepairJobDesc desc, - List trees, - InetAddressAndPort local, - Predicate isTransient, - Function getDC, - boolean isIncremental, - PreviewKind previewKind, + @SuppressWarnings("unused") + public static List createOptimisedSyncingSyncTasks(List trees, @SuperCall Callable> zuperCall) { - List tasks = null; + List tasks; try { tasks = zuperCall.call(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/ConcurrentValidationRequestsTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/ConcurrentValidationRequestsTest.java new file mode 100644 index 0000000000..82a0b73371 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/ConcurrentValidationRequestsTest.java @@ -0,0 +1,136 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test.repair; + +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; +import net.bytebuddy.implementation.bind.annotation.SuperCall; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.NodeToolResult; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.repair.ValidationTask; +import org.apache.cassandra.utils.MerkleTrees; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + +/** + * Verifies that the config property {@code concurrent_merkle_tree_requests} limits the number of concurrent validation + * requests. + */ +public class ConcurrentValidationRequestsTest extends TestBaseImpl +{ + private static final int NODES = 4; + private static final int RF = 3; + private static final int TABLES = 5; + private static final int ROWS = 100; + private static final int MAX_REQUESTS = 1; + + @Test + public void testConcurrentValidations() throws Throwable + { + try (Cluster cluster = init(builder().withNodes(NODES) + .withInstanceInitializer(ConcurrentValidationRequestsTest.BBHelper::install) + .withConfig(c -> c.set("concurrent_merkle_tree_requests", MAX_REQUESTS) + .with(NETWORK, GOSSIP)) + .start(), RF)) + { + for (int t = 1; t <= TABLES; t++) + cluster.schemaChange(withKeyspace("CREATE TABLE %s.t" + t + " (k int PRIMARY KEY, v int)")); + + int v = 0; + for (int k = 0; k < ROWS; k++) + { + v++; + for (int t = 1; t <= TABLES; t++) + { + String insert = withKeyspace("INSERT INTO %s.t" + t + " (k, v) VALUES (?, ?)"); + cluster.coordinator(1).execute(insert, ConsistencyLevel.ALL, k, v); + } + } + cluster.forEach(x -> x.flush(KEYSPACE)); + + NodeToolResult res = cluster.get(1).nodetoolResult("repair", "-j=4", KEYSPACE); + res.asserts().success(); + } + } + + public static class BBHelper + { + /** + * Keeps track of the number of concurrent validation requests, which should never be greater than RF times + * the value of the {@code concurrent_merkle_tree_requests} config property. + */ + public static volatile AtomicInteger requests = new AtomicInteger(0); + + public static void install(ClassLoader cl, int node) + { + if (node != 1) + return; + + new ByteBuddy().rebase(ValidationTask.class) + .method(named("run")) + .intercept(MethodDelegation.to(ConcurrentValidationRequestsTest.BBHelper.class)) + .method(named("treesReceived")) + .intercept(MethodDelegation.to(ConcurrentValidationRequestsTest.BBHelper.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + + @SuppressWarnings("unused") + public static void run(@SuperCall Callable zuper) + { + try + { + zuper.call(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + + int requests = BBHelper.requests.incrementAndGet(); + if (requests > MAX_REQUESTS * RF) + throw new AssertionError("Too many concurrent validation requests: " + requests); + } + + @SuppressWarnings("unused") + public static void treesReceived(MerkleTrees trees, @SuperCall Callable zuperCall) + { + requests.decrementAndGet(); + + try + { + zuperCall.call(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + } + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/repair/RepairJobTest.java b/test/unit/org/apache/cassandra/repair/RepairJobTest.java index c7c68c4689..b164b4f2e8 100644 --- a/test/unit/org/apache/cassandra/repair/RepairJobTest.java +++ b/test/unit/org/apache/cassandra/repair/RepairJobTest.java @@ -127,8 +127,8 @@ public class RepairJobTest PreviewKind previewKind, boolean optimiseStreams, boolean repairPaxos, boolean paxosOnly, String... cfnames) { - super(parentRepairSession, commonRange, keyspace, parallelismDegree, isIncremental, pullRepair, - previewKind, optimiseStreams, repairPaxos, paxosOnly, cfnames); + super(parentRepairSession, new Scheduler.NoopScheduler(), commonRange, keyspace, parallelismDegree, + isIncremental, pullRepair, previewKind, optimiseStreams, repairPaxos, paxosOnly, cfnames); } protected ExecutorPlus createExecutor() @@ -332,7 +332,7 @@ public class RepairJobTest } catch (ExecutionException e) { - Assertions.assertThat(e.getCause()).isInstanceOf(RepairException.class); + Assertions.assertThat(e).hasRootCauseInstanceOf(RepairException.class); } // When the job fails, all three outstanding validation tasks should be aborted. diff --git a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java index 4db6efb680..88eff6ca0a 100644 --- a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java @@ -63,7 +63,7 @@ public class RepairSessionTest IPartitioner p = Murmur3Partitioner.instance; Range repairRange = new Range<>(p.getToken(ByteBufferUtil.bytes(0)), p.getToken(ByteBufferUtil.bytes(100))); Set endpoints = Sets.newHashSet(remote); - RepairSession session = new RepairSession(parentSessionId, + RepairSession session = new RepairSession(parentSessionId, new Scheduler.NoopScheduler(), new CommonRange(endpoints, Collections.emptySet(), Arrays.asList(repairRange)), "Keyspace1", RepairParallelism.SEQUENTIAL, false, false,