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

Merge branch 'cassandra-4.1' into cassandra-5.0

This commit is contained in:
Andrés de la Peña 2024-02-08 11:33:23 +00:00
parent 78dca99973 89a8155916
commit 5b9321eee1
19 changed files with 412 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -18,6 +18,7 @@ Merged from 4.1:
* Memoize Cassandra verion and add a backoff interval for failed schema pulls (CASSANDRA-18902)
* Fix StackOverflowError on ALTER after many previous schema changes (CASSANDRA-19166)
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)

36
conf/cassandra.yaml
View File

@ -738,18 +738,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
# repair:
# # Configure the retries for each of the repair messages that support it. As of this moment retries use an exponential algorithm where each attempt sleeps longer based off the base_sleep_time and attempt.
# retries:

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

@ -205,6 +205,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;

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

@ -3859,6 +3859,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;

5
src/java/org/apache/cassandra/repair/AbstractRepairTask.java
View File

@ -57,6 +57,7 @@ public abstract class AbstractRepairTask implements RepairTask
private List<RepairSession> submitRepairSessions(TimeUUID parentSession,
boolean isIncremental,
ExecutorPlus executor,
Scheduler validationScheduler,
List<CommonRange> commonRanges,
String... cfnames)
{
@ -76,6 +77,7 @@ public abstract class AbstractRepairTask implements RepairTask
options.repairPaxos(),
options.paxosOnly(),
executor,
validationScheduler,
cfnames);
if (session == null)
continue;
@ -88,10 +90,11 @@ public abstract class AbstractRepairTask implements RepairTask
protected Future<CoordinatedRepairResult> runRepair(TimeUUID parentSession,
boolean isIncremental,
ExecutorPlus executor,
Scheduler validationScheduler,
List<CommonRange> commonRanges,
String... cfnames)
{
List<RepairSession> allSessions = submitRepairSessions(parentSession, isIncremental, executor, commonRanges, cfnames);
List<RepairSession> allSessions = submitRepairSessions(parentSession, isIncremental, executor, validationScheduler, commonRanges, cfnames);
List<Collection<Range<Token>>> ranges = Lists.transform(allSessions, RepairSession::ranges);
Future<List<RepairSessionResult>> f = FutureCombiner.successfulOf(allSessions);
return f.map(results -> {

4
src/java/org/apache/cassandra/repair/IncrementalRepairTask.java
View File

@ -52,7 +52,7 @@ public class IncrementalRepairTask extends AbstractRepairTask
}
@Override
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor) throws Exception
public Future<CoordinatedRepairResult> 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<InetAddressAndPort> allParticipants = ImmutableSet.<InetAddressAndPort>builder()
@ -64,7 +64,7 @@ public class IncrementalRepairTask extends AbstractRepairTask
CoordinatorSession coordinatorSession = coordinator.ctx.repair().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));
}
}

4
src/java/org/apache/cassandra/repair/NormalRepairTask.java
View File

@ -47,8 +47,8 @@ public class NormalRepairTask extends AbstractRepairTask
}
@Override
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor)
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor, Scheduler validationScheduler)
{
return runRepair(parentSession, false, executor, commonRanges, cfnames);
return runRepair(parentSession, false, executor, validationScheduler, commonRanges, cfnames);
}
}

4
src/java/org/apache/cassandra/repair/PreviewRepairTask.java
View File

@ -65,9 +65,9 @@ public class PreviewRepairTask extends AbstractRepairTask
}
@Override
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor)
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor, Scheduler validationScheduler)
{
Future<CoordinatedRepairResult> f = runRepair(parentSession, false, executor, commonRanges, cfnames);
Future<CoordinatedRepairResult> f = runRepair(parentSession, false, executor, validationScheduler, commonRanges, cfnames);
return f.map(result -> {
if (result.hasFailed())
return result;

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

@ -99,6 +99,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
private final List<ProgressListener> listeners = new ArrayList<>();
private final AtomicReference<Throwable> firstError = new AtomicReference<>(null);
final SharedContext ctx;
final Scheduler validationScheduler;
private TraceState traceState;
@ -116,6 +117,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
int cmd, RepairOption options, String keyspace)
{
this.ctx = ctx;
this.validationScheduler = Scheduler.build(DatabaseDescriptor.getConcurrentMerkleTreeRequests());
this.state = new CoordinatorState(ctx.clock(), cmd, keyspace, options);
this.tag = "repair:" + cmd;
this.validColumnFamilies = validColumnFamilies;
@ -453,7 +455,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
ExecutorPlus executor = createExecutor();
state.phase.repairSubmitted();
return task.perform(executor)
return task.perform(executor, validationScheduler)
// after adding the callback java could no longer infer the type...
.<Pair<CoordinatedRepairResult, Supplier<String>>>map(r -> Pair.create(r, task::successMessage))
.addCallback((s, f) -> executor.shutdown());

129
src/java/org/apache/cassandra/repair/RepairJob.java
View File

@ -20,11 +20,9 @@ package org.apache.cassandra.repair;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.function.Predicate;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
@ -42,16 +40,16 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.repair.asymmetric.DifferenceHolder;
import org.apache.cassandra.repair.asymmetric.HostDifferences;
import org.apache.cassandra.repair.asymmetric.PreferedNodeFilter;
import org.apache.cassandra.repair.asymmetric.ReduceHelper;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.SystemDistributedKeyspace;
import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.paxos.cleanup.PaxosCleanup;
import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTrees;
@ -113,7 +111,7 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
/**
* Runs repair job.
*
* <p/>
* This sets up necessary task and runs them on given {@code taskExecutor}.
* After submitting all tasks, waits until validation with replica completes.
*/
@ -126,7 +124,6 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
List<InetAddressAndPort> allEndpoints = new ArrayList<>(session.state.commonRange.endpoints);
allEndpoints.add(ctx.broadcastAddressAndPort());
Future<List<TreeResponse>> treeResponses;
Future<Void> paxosRepair;
if (paxosRepairEnabled() && ((useV2() && session.repairPaxos) || session.paxosOnly))
{
@ -142,7 +139,7 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
if (session.paxosOnly)
{
paxosRepair.addCallback(new FutureCallback<Void>()
paxosRepair.addCallback(new FutureCallback<>()
{
public void onSuccess(Void v)
{
@ -163,9 +160,9 @@ public class RepairJob extends AsyncFuture<RepairResult> 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"
@ -189,30 +186,19 @@ public class RepairJob extends AsyncFuture<RepairResult> 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<List<SyncStat>> 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<List<SyncStat>> 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<List<SyncStat>>()
syncResults.addCallback(new FutureCallback<>()
{
@Override
public void onSuccess(List<SyncStat> stats)
@ -249,6 +235,35 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
}, taskExecutor);
}
private Future<List<SyncTask>> createSyncTasks(Future<Void> paxosRepair, Future<?> allSnapshotTasks, List<InetAddressAndPort> allEndpoints)
{
Future<List<TreeResponse>> 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);
}
public synchronized void abort(@Nullable Throwable reason)
{
if (reason == null)
@ -265,18 +280,18 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
return session.state.commonRange.transEndpoints.contains(ep);
}
private Future<List<SyncStat>> standardSyncing(List<TreeResponse> trees)
private List<SyncTask> createStandardSyncTasks(List<TreeResponse> trees)
{
List<SyncTask> syncTasks = createStandardSyncTasks(ctx, desc,
trees,
ctx.broadcastAddressAndPort(),
this::isTransient,
session.isIncremental,
session.pullRepair,
session.previewKind);
return executeTasks(syncTasks);
return createStandardSyncTasks(ctx, desc,
trees,
ctx.broadcastAddressAndPort(),
this::isTransient,
session.isIncremental,
session.pullRepair,
session.previewKind);
}
@VisibleForTesting
static List<SyncTask> createStandardSyncTasks(SharedContext ctx,
RepairJobDesc desc,
List<TreeResponse> trees,
@ -345,20 +360,6 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
return syncTasks;
}
private Future<List<SyncStat>> optimisedSyncing(List<TreeResponse> trees)
{
List<SyncTask> syncTasks = createOptimisedSyncingSyncTasks(ctx,
desc,
trees,
FBUtilities.getLocalAddressAndPort(),
this::isTransient,
this::getDC,
session.isIncremental,
session.previewKind);
return executeTasks(syncTasks);
}
@VisibleForTesting
Future<List<SyncStat>> executeTasks(List<SyncTask> tasks)
{
@ -400,6 +401,19 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
}
}
private List<SyncTask> createOptimisedSyncingSyncTasks(List<TreeResponse> trees)
{
return createOptimisedSyncingSyncTasks(ctx,
desc,
trees,
FBUtilities.getLocalAddressAndPort(),
this::isTransient,
this::getDC,
session.isIncremental,
session.previewKind);
}
@VisibleForTesting
static List<SyncTask> createOptimisedSyncingSyncTasks(SharedContext ctx,
RepairJobDesc desc,
List<TreeResponse> trees,
@ -483,7 +497,7 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
logger.info("{} {}", session.previewKind.logPrefix(desc.sessionId), message);
Tracing.traceRepair(message);
long nowInSec = getNowInSeconds();
List<Future<TreeResponse>> tasks = new ArrayList<>(endpoints.size());
List<ValidationTask> tasks = new ArrayList<>(endpoints.size());
for (InetAddressAndPort endpoint : endpoints)
{
ValidationTask task = newValidationTask(endpoint, nowInSec);
@ -518,7 +532,7 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
final InetAddressAndPort nextAddress = requests.poll();
final ValidationTask nextTask = newValidationTask(nextAddress, nowInSec);
tasks.add(nextTask);
currentTask.addCallback(new FutureCallback<TreeResponse>()
currentTask.addCallback(new FutureCallback<>()
{
public void onSuccess(TreeResponse result)
{
@ -553,12 +567,7 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
for (InetAddressAndPort endpoint : endpoints)
{
String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint);
Queue<InetAddressAndPort> queue = requestsByDatacenter.get(dc);
if (queue == null)
{
queue = new LinkedList<>();
requestsByDatacenter.put(dc, queue);
}
Queue<InetAddressAndPort> queue = requestsByDatacenter.computeIfAbsent(dc, k -> new LinkedList<>());
queue.add(endpoint);
}
@ -576,7 +585,7 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
final InetAddressAndPort nextAddress = requests.poll();
final ValidationTask nextTask = newValidationTask(nextAddress, nowInSec);
tasks.add(nextTask);
currentTask.addCallback(new FutureCallback<TreeResponse>()
currentTask.addCallback(new FutureCallback<>()
{
public void onSuccess(TreeResponse result)
{

3
src/java/org/apache/cassandra/repair/RepairSession.java
View File

@ -133,6 +133,7 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
public final SafeExecutor taskExecutor;
public final boolean optimiseStreams;
public final SharedContext ctx;
public final Scheduler validationScheduler;
private volatile List<RepairJob> jobs = Collections.emptyList();
private volatile boolean terminated = false;
@ -149,6 +150,7 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
* @param cfnames names of columnfamilies
*/
public RepairSession(SharedContext ctx,
Scheduler validationScheduler,
TimeUUID parentRepairSession,
CommonRange commonRange,
String keyspace,
@ -162,6 +164,7 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
String... cfnames)
{
this.ctx = ctx;
this.validationScheduler = validationScheduler;
this.repairPaxos = repairPaxos;
this.paxosOnly = paxosOnly;
assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it";

6
src/java/org/apache/cassandra/repair/RepairTask.java
View File

@ -31,13 +31,13 @@ public interface RepairTask
return name() + " completed successfully";
}
Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor) throws Exception;
Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor, Scheduler validationScheduler) throws Exception;
default Future<CoordinatedRepairResult> perform(ExecutorPlus executor)
default Future<CoordinatedRepairResult> perform(ExecutorPlus executor, Scheduler validationScheduler)
{
try
{
return performUnsafe(executor);
return performUnsafe(executor, validationScheduler);
}
catch (Exception | Error e)
{

118
src/java/org/apache/cassandra/repair/Scheduler.java Normal file
View File

@ -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 <T> Future<T> schedule(Supplier<Future<T>> task, Executor executor)
{
return schedule(new Task<>(task), executor);
}
<T> Task<T> schedule(Task<T> task, Executor executor);
static Scheduler build(int concurrentValidations)
{
return concurrentValidations <= 0
? new NoopScheduler()
: new LimitedConcurrentScheduler(concurrentValidations);
}
final class NoopScheduler implements Scheduler
{
@Override
public <T> Task<T> schedule(Task<T> 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<Pair<Task<?>, Executor>> tasks = new LinkedList<>();
LimitedConcurrentScheduler(int concurrentValidations)
{
this.concurrentValidations = concurrentValidations;
}
@Override
public synchronized <T> Task<T> schedule(Task<T> 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<Task<?>, Executor> pair = tasks.poll();
pair.left.addCallback((s, f) -> onDone());
pair.right.execute(pair.left);
}
}
class Task<T> extends AsyncFuture<T> implements Runnable
{
private final Supplier<Future<T>> supplier;
public Task(Supplier<Future<T>> supplier)
{
this.supplier = supplier;
}
@Override
public void run()
{
supplier.get().addCallback((s, f) -> {
if (f != null)
tryFailure(f);
else
trySuccess(s);
});
}
}
}

17
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -44,6 +44,7 @@ import com.google.common.collect.Multimap;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.repair.Scheduler;
import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
@ -351,6 +352,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<CompositeData> getRepairStats(List<String> schemaArgs, String rangeString)
{
List<CompositeData> stats = new ArrayList<>();
@ -432,6 +446,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
boolean repairPaxos,
boolean paxosOnly,
ExecutorPlus executor,
Scheduler validationScheduler,
String... cfnames)
{
if (repairPaxos && previewKind != PreviewKind.NONE)
@ -443,7 +458,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
if (cfnames.length == 0)
return null;
final RepairSession session = new RepairSession(ctx, parentRepairSession, range, keyspace,
final RepairSession session = new RepairSession(ctx, validationScheduler, parentRepairSession, range, keyspace,
parallelismDegree, isIncremental, pullRepair,
previewKind, optimiseStreams, repairPaxos, paxosOnly, cfnames);
repairs.getIfPresent(parentRepairSession).register(session.state);

4
src/java/org/apache/cassandra/service/ActiveRepairServiceMBean.java
View File

@ -50,6 +50,10 @@ public interface ActiveRepairServiceMBean
public void setRepairSessionSpaceInMiB(int sizeInMebibytes);
public int getRepairSessionSpaceInMiB();
int getConcurrentMerkleTreeRequests();
void setConcurrentMerkleTreeRequests(int value);
public boolean getUseOffheapMerkleTrees();
public void setUseOffheapMerkleTrees(boolean value);

17
test/distributed/org/apache/cassandra/distributed/test/OptimiseStreamsRepairTest.java
View File

@ -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;
@ -39,7 +37,6 @@ 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.repair.SharedContext;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.distributed.Cluster;
@ -49,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;
@ -109,20 +105,13 @@ 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<SyncTask> createOptimisedSyncingSyncTasks(SharedContext ctx,
RepairJobDesc desc,
List<TreeResponse> trees,
InetAddressAndPort local,
Predicate<InetAddressAndPort> isTransient,
Function<InetAddressAndPort, String> getDC,
boolean isIncremental,
PreviewKind previewKind,
public static List<SyncTask> createOptimisedSyncingSyncTasks(List<TreeResponse> trees,
@SuperCall Callable<List<SyncTask>> zuperCall)
{
List<SyncTask> tasks = null;

136
test/distributed/org/apache/cassandra/distributed/test/repair/ConcurrentValidationRequestsTest.java Normal file
View File

@ -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<Void> 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<Void> zuperCall)
{
requests.decrementAndGet();
try
{
zuperCall.call();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
}
}

5
test/unit/org/apache/cassandra/repair/RepairJobTest.java
View File

@ -127,7 +127,8 @@ public class RepairJobTest
PreviewKind previewKind, boolean optimiseStreams, boolean repairPaxos, boolean paxosOnly,
String... cfnames)
{
super(SharedContext.Global.instance, parentRepairSession, commonRange, keyspace, parallelismDegree, isIncremental, pullRepair,
super(SharedContext.Global.instance, new Scheduler.NoopScheduler(),
parentRepairSession, commonRange, keyspace, parallelismDegree, isIncremental, pullRepair,
previewKind, optimiseStreams, repairPaxos, paxosOnly, cfnames);
}
@ -333,7 +334,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.

2
test/unit/org/apache/cassandra/repair/RepairSessionTest.java
View File

@ -63,7 +63,7 @@ public class RepairSessionTest
IPartitioner p = Murmur3Partitioner.instance;
Range<Token> repairRange = new Range<>(p.getToken(ByteBufferUtil.bytes(0)), p.getToken(ByteBufferUtil.bytes(100)));
Set<InetAddressAndPort> endpoints = Sets.newHashSet(remote);
RepairSession session = new RepairSession(SharedContext.Global.instance, parentSessionId,
RepairSession session = new RepairSession(SharedContext.Global.instance, new Scheduler.NoopScheduler(), parentSessionId,
new CommonRange(endpoints, Collections.emptySet(), Arrays.asList(repairRange)),
"Keyspace1", RepairParallelism.SEQUENTIAL,
false, false,