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

improve concurrency of repair process 

patch by yukim; reviewed by krummas for CASSANDRA-6455
This commit is contained in:
Yuki Morishita 2014-10-09 18:12:36 -05:00
parent 5c35f9203f
commit 810c2d5fe6
28 changed files with 1459 additions and 1213 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -29,6 +29,7 @@
* Use unsafe mutations for most unit tests (CASSANDRA-6969)
* Fix race condition during calculation of pending ranges (CASSANDRA-7390)
* Fail on very large batch sizes (CASSANDRA-8011)
* improve concurrency of repair (CASSANDRA-6455)
2.1.1

136
src/java/org/apache/cassandra/repair/Differencer.java
View File

@ -1,136 +0,0 @@
/*
* 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.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import com.google.common.base.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.SyncComplete;
import org.apache.cassandra.repair.messages.SyncRequest;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
/**
* Runs on the node that initiated a request to compare two trees, and launch repairs for disagreeing ranges.
*/
public class Differencer implements Runnable
{
private static Logger logger = LoggerFactory.getLogger(Differencer.class);
private final RepairJobDesc desc;
public final TreeResponse r1;
public final TreeResponse r2;
public final List<Range<Token>> differences = new ArrayList<>();
public Differencer(RepairJobDesc desc, TreeResponse r1, TreeResponse r2)
{
this.desc = desc;
this.r1 = r1;
this.r2 = r2;
}
/**
* Compares our trees, and triggers repairs for any ranges that mismatch.
*/
public void run()
{
// compare trees, and collect differences
differences.addAll(MerkleTree.difference(r1.tree, r2.tree));
// choose a repair method based on the significance of the difference
String format = String.format("[repair #%s] Endpoints %s and %s %%s for %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily);
if (differences.isEmpty())
{
logger.info(String.format(format, "are consistent"));
// send back sync complete message
MessagingService.instance().sendOneWay(new SyncComplete(desc, r1.endpoint, r2.endpoint, true).createMessage(), FBUtilities.getLocalAddress());
return;
}
// non-0 difference: perform streaming repair
logger.info(String.format(format, "have " + differences.size() + " range(s) out of sync"));
performStreamingRepair();
}
/**
* Starts sending/receiving our list of differences to/from the remote endpoint: creates a callback
* that will be called out of band once the streams complete.
*/
void performStreamingRepair()
{
InetAddress local = FBUtilities.getBroadcastAddress();
// We can take anyone of the node as source or destination, however if one is localhost, we put at source to avoid a forwarding
InetAddress src = r2.endpoint.equals(local) ? r2.endpoint : r1.endpoint;
InetAddress dst = r2.endpoint.equals(local) ? r1.endpoint : r2.endpoint;
SyncRequest request = new SyncRequest(desc, local, src, dst, differences);
StreamingRepairTask task = new StreamingRepairTask(desc, request);
task.run();
}
/**
* In order to remove completed Differencer, equality is computed only from {@code desc} and
* endpoint part of two TreeResponses.
*/
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Differencer that = (Differencer) o;
if (!desc.equals(that.desc)) return false;
return minEndpoint().equals(that.minEndpoint()) && maxEndpoint().equals(that.maxEndpoint());
}
@Override
public int hashCode()
{
return Objects.hashCode(desc, minEndpoint(), maxEndpoint());
}
// For equals and hashcode, we don't want to take the endpoint order into account.
// So we just order endpoint deterministically to simplify this
private InetAddress minEndpoint()
{
return FBUtilities.compareUnsigned(r1.endpoint.getAddress(), r2.endpoint.getAddress()) < 0
? r1.endpoint
: r2.endpoint;
}
private InetAddress maxEndpoint()
{
return FBUtilities.compareUnsigned(r1.endpoint.getAddress(), r2.endpoint.getAddress()) < 0
? r2.endpoint
: r1.endpoint;
}
public String toString()
{
return "#<Differencer " + r1.endpoint + "<->" + r2.endpoint + ":" + desc.columnFamily + "@" + desc.range + ">";
}
}

81
src/java/org/apache/cassandra/repair/LocalSyncTask.java Normal file
View File

@ -0,0 +1,81 @@
/*
* 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.net.InetAddress;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.streaming.StreamEvent;
import org.apache.cassandra.streaming.StreamEventHandler;
import org.apache.cassandra.streaming.StreamPlan;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.utils.FBUtilities;
/**
* LocalSyncTask performs streaming between local(coordinator) node and remote replica.
*/
public class LocalSyncTask extends SyncTask implements StreamEventHandler
{
private static final Logger logger = LoggerFactory.getLogger(LocalSyncTask.class);
private final long repairedAt;
public LocalSyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2, long repairedAt)
{
super(desc, r1, r2);
this.repairedAt = repairedAt;
}
/**
* Starts sending/receiving our list of differences to/from the remote endpoint: creates a callback
* that will be called out of band once the streams complete.
*/
protected void startSync(List<Range<Token>> differences)
{
InetAddress local = FBUtilities.getBroadcastAddress();
// We can take anyone of the node as source or destination, however if one is localhost, we put at source to avoid a forwarding
InetAddress dst = r2.endpoint.equals(local) ? r1.endpoint : r2.endpoint;
logger.info(String.format("[repair #%s] Performing streaming repair of %d ranges with %s", desc.sessionId, differences.size(), dst));
new StreamPlan("Repair", repairedAt, 1).listeners(this)
.flushBeforeTransfer(true)
// request ranges from the remote node
.requestRanges(dst, desc.keyspace, differences, desc.columnFamily)
// send ranges to the remote node
.transferRanges(dst, desc.keyspace, differences, desc.columnFamily)
.execute();
}
public void handleStreamEvent(StreamEvent event) { /* noop */ }
public void onSuccess(StreamState result)
{
logger.info(String.format("[repair #%s] Sync complete between %s and %s on %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily));
set(stat);
}
public void onFailure(Throwable t)
{
setException(t);
}
}

67
src/java/org/apache/cassandra/repair/RemoteSyncTask.java Normal file
View File

@ -0,0 +1,67 @@
/*
* 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.net.InetAddress;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RepairException;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.SyncRequest;
import org.apache.cassandra.utils.FBUtilities;
/**
* RemoteSyncTask sends {@link SyncRequest} to remote(non-coordinator) node
* to repair(stream) data with other replica.
*
* When RemoteSyncTask receives SyncComplete from remote node, task completes.
*/
public class RemoteSyncTask extends SyncTask
{
private static final Logger logger = LoggerFactory.getLogger(RemoteSyncTask.class);
public RemoteSyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2)
{
super(desc, r1, r2);
}
protected void startSync(List<Range<Token>> differences)
{
InetAddress local = FBUtilities.getBroadcastAddress();
SyncRequest request = new SyncRequest(desc, local, r1.endpoint, r2.endpoint, differences);
logger.info(String.format("[repair #%s] Forwarding streaming repair of %d ranges to %s (to be streamed with %s)", desc.sessionId, request.ranges.size(), request.src, request.dst));
MessagingService.instance().sendOneWay(request.createMessage(), request.src);
}
public void syncComplete(boolean success)
{
if (success)
{
set(stat);
}
else
{
setException(new RepairException(desc, String.format("Sync failed between %s and %s", r1.endpoint, r2.endpoint)));
}
}
}

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

@ -18,93 +18,68 @@
package org.apache.cassandra.repair;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import com.google.common.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.ValidationRequest;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.apache.cassandra.utils.Pair;
/**
* RepairJob runs repair on given ColumnFamily.
*/
public class RepairJob
public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
{
private static Logger logger = LoggerFactory.getLogger(RepairJob.class);
public final RepairJobDesc desc;
private final RepairSession session;
private final RepairJobDesc desc;
private final boolean isSequential;
// first we send tree requests. this tracks the endpoints remaining to hear from
private final RequestCoordinator<InetAddress> treeRequests;
// tree responses are then tracked here
private final List<TreeResponse> trees = new ArrayList<>();
// once all responses are received, each tree is compared with each other, and differencer tasks
// are submitted. the job is done when all differencers are complete.
private final long repairedAt;
private final ListeningExecutorService taskExecutor;
private final Condition requestsSent = new SimpleCondition();
private int gcBefore = -1;
private volatile boolean failed = false;
/* Count down as sync completes */
private AtomicInteger waitForSync;
private final IRepairJobEventListener listener;
/**
* Create repair job to run on specific columnfamily
*
* @param session RepairSession that this RepairJob belongs
* @param columnFamily name of the ColumnFamily to repair
* @param isSequential when true, validation runs sequentially among replica
* @param taskExecutor Executor to run various repair tasks
*/
public RepairJob(IRepairJobEventListener listener,
UUID parentSessionId,
UUID sessionId,
String keyspace,
public RepairJob(RepairSession session,
String columnFamily,
Range<Token> range,
boolean isSequential,
long repairedAt,
ListeningExecutorService taskExecutor)
{
this.listener = listener;
this.desc = new RepairJobDesc(parentSessionId, sessionId, keyspace, columnFamily, range);
this.session = session;
this.desc = new RepairJobDesc(session.parentRepairSession, session.getId(), session.keyspace, columnFamily, session.getRange());
this.isSequential = isSequential;
this.repairedAt = repairedAt;
this.taskExecutor = taskExecutor;
this.treeRequests = new RequestCoordinator<InetAddress>(isSequential)
{
public void send(InetAddress endpoint)
{
ValidationRequest request = new ValidationRequest(desc, gcBefore);
MessagingService.instance().sendOneWay(request.createMessage(), endpoint);
}
};
}
/**
* @return true if this job failed
* 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.
*/
public boolean isFailed()
public void run()
{
return failed;
}
/**
* Send merkle tree request to every involved neighbor.
*/
public void sendTreeRequests(Collection<InetAddress> endpoints)
{
// send requests to all nodes
List<InetAddress> allEndpoints = new ArrayList<>(endpoints);
List<InetAddress> allEndpoints = new ArrayList<>(session.endpoints);
allEndpoints.add(FBUtilities.getBroadcastAddress());
ListenableFuture<List<TreeResponse>> validations;
if (isSequential)
{
// Request snapshot to all replica
List<ListenableFuture<InetAddress>> snapshotTasks = new ArrayList<>(allEndpoints.size());
for (InetAddress endpoint : allEndpoints)
{
@ -112,102 +87,110 @@ public class RepairJob
snapshotTasks.add(snapshotTask);
taskExecutor.execute(snapshotTask);
}
// When all snapshot complete, send validation requests
ListenableFuture<List<InetAddress>> allSnapshotTasks = Futures.allAsList(snapshotTasks);
// Execute send tree request after all snapshot complete
Futures.addCallback(allSnapshotTasks, new FutureCallback<List<InetAddress>>()
validations = Futures.transform(allSnapshotTasks, new AsyncFunction<List<InetAddress>, List<TreeResponse>>()
{
public void onSuccess(List<InetAddress> endpoints)
public ListenableFuture<List<TreeResponse>> apply(List<InetAddress> endpoints) throws Exception
{
sendTreeRequestsInternal(endpoints);
}
public void onFailure(Throwable throwable)
{
// TODO need to propagate error to RepairSession
logger.error("Error occurred during snapshot phase", throwable);
listener.failedSnapshot();
failed = true;
return sendValidationRequest(endpoints);
}
}, taskExecutor);
}
else
{
sendTreeRequestsInternal(allEndpoints);
}
}
private void sendTreeRequestsInternal(Collection<InetAddress> endpoints)
{
this.gcBefore = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily).gcBefore(System.currentTimeMillis());
for (InetAddress endpoint : endpoints)
treeRequests.add(endpoint);
logger.info(String.format("[repair #%s] requesting merkle trees for %s (to %s)", desc.sessionId, desc.columnFamily, endpoints));
treeRequests.start();
requestsSent.signalAll();
}
/**
* Add a new received tree and return the number of remaining tree to
* be received for the job to be complete.
*
* Callers may assume exactly one addTree call will result in zero remaining endpoints.
*
* @param endpoint address of the endpoint that sent response
* @param tree sent Merkle tree or null if validation failed on endpoint
* @return the number of responses waiting to receive
*/
public synchronized int addTree(InetAddress endpoint, MerkleTree tree)
{
// Wait for all request to have been performed (see #3400)
try
{
requestsSent.await();
}
catch (InterruptedException e)
{
throw new AssertionError("Interrupted while waiting for requests to be sent");
// If not sequential, just send validation request to all replica
validations = sendValidationRequest(allEndpoints);
}
if (tree == null)
failed = true;
else
trees.add(new TreeResponse(endpoint, tree));
return treeRequests.completed(endpoint);
}
/**
* Submit differencers for running.
* All tree *must* have been received before this is called.
*/
public void submitDifferencers()
{
assert !failed;
List<Differencer> differencers = new ArrayList<>();
// We need to difference all trees one against another
for (int i = 0; i < trees.size() - 1; ++i)
// When all validations complete, submit sync tasks
ListenableFuture<List<SyncStat>> syncResults = Futures.transform(validations, new AsyncFunction<List<TreeResponse>, List<SyncStat>>()
{
TreeResponse r1 = trees.get(i);
for (int j = i + 1; j < trees.size(); ++j)
public ListenableFuture<List<SyncStat>> apply(List<TreeResponse> trees) throws Exception
{
TreeResponse r2 = trees.get(j);
Differencer differencer = new Differencer(desc, r1, r2);
differencers.add(differencer);
logger.debug("Queueing comparison {}", differencer);
// Unregister from FailureDetector once we've completed synchronizing Merkle trees.
// After this point, we rely on tcp_keepalive for individual sockets to notify us when a connection is down.
// See CASSANDRA-3569
FailureDetector.instance.unregisterFailureDetectionEventListener(session);
InetAddress local = FBUtilities.getLocalAddress();
List<SyncTask> syncTasks = new ArrayList<>();
// We need to difference all trees one against another
for (int i = 0; i < trees.size() - 1; ++i)
{
TreeResponse r1 = trees.get(i);
for (int j = i + 1; j < trees.size(); ++j)
{
TreeResponse r2 = trees.get(j);
SyncTask task;
if (r1.endpoint.equals(local) || r2.endpoint.equals(local))
{
task = new LocalSyncTask(desc, r1, r2, repairedAt);
}
else
{
task = new RemoteSyncTask(desc, r1, r2);
// RemoteSyncTask expects SyncComplete message sent back.
// Register task to RepairSession to receive response.
session.waitForSync(Pair.create(desc, new NodePair(r1.endpoint, r2.endpoint)), (RemoteSyncTask) task);
}
syncTasks.add(task);
taskExecutor.submit(task);
}
}
return Futures.allAsList(syncTasks);
}
}, taskExecutor);
// When all sync complete, set the final result
Futures.addCallback(syncResults, new FutureCallback<List<SyncStat>>()
{
public void onSuccess(List<SyncStat> stats)
{
logger.info(String.format("[repair #%s] %s is fully synced", session.getId(), desc.columnFamily));
set(new RepairResult(desc, stats));
}
/**
* Snapshot, validation and sync failures are all handled here
*/
public void onFailure(Throwable t)
{
logger.warn(String.format("[repair #%s] %s sync failed", session.getId(), desc.columnFamily));
setException(t);
}
}, taskExecutor);
// Wait for validation to complete
Futures.getUnchecked(validations);
}
/**
* Creates {@link ValidationTask} and submit them to task executor.
* If isSequential flag is true, wait previous ValidationTask to complete before submitting the next.
*
* @param endpoints Endpoint addresses to send validation request
* @return Future that can get all {@link TreeResponse} from replica, if all validation succeed.
*/
private ListenableFuture<List<TreeResponse>> sendValidationRequest(Collection<InetAddress> endpoints)
{
logger.info(String.format("[repair #%s] requesting merkle trees for %s (to %s)", desc.sessionId, desc.columnFamily, endpoints));
int gcBefore = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily).gcBefore(System.currentTimeMillis());
List<ListenableFuture<TreeResponse>> tasks = new ArrayList<>(endpoints.size());
for (InetAddress endpoint : endpoints)
{
ValidationTask task = new ValidationTask(desc, endpoint, gcBefore);
tasks.add(task);
session.waitForValidation(Pair.create(desc, endpoint), task);
taskExecutor.execute(task);
if (isSequential)
{
// tasks are sequentially sent so wait until current validation is done.
// NOTE: Wait happens on taskExecutor thread
Futures.getUnchecked(task);
}
}
waitForSync = new AtomicInteger(differencers.size());
for (Differencer differencer : differencers)
taskExecutor.submit(differencer);
trees.clear(); // allows gc to do its thing
}
/**
* @return true if the given node pair was the last remaining
*/
boolean completedSynchronization()
{
return waitForSync.decrementAndGet() == 0;
return Futures.allAsList(tasks);
}
}

7
src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java
View File

@ -103,7 +103,12 @@ public class RepairMessageVerbHandler implements IVerbHandler<RepairMessage>
case SYNC_REQUEST:
// forwarded sync request
SyncRequest request = (SyncRequest) message.payload;
StreamingRepairTask task = new StreamingRepairTask(desc, request);
long repairedAt = ActiveRepairService.UNREPAIRED_SSTABLE;
if (desc.parentSessionId != null && ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId) != null)
repairedAt = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).repairedAt;
StreamingRepairTask task = new StreamingRepairTask(desc, request, repairedAt);
task.run();
break;

13
src/java/org/apache/cassandra/repair/RepairFuture.java → src/java/org/apache/cassandra/repair/RepairResult.java
View File

@ -17,15 +17,16 @@
*/
package org.apache.cassandra.repair;
import java.util.concurrent.FutureTask;
import java.util.List;
public class RepairFuture extends FutureTask<Void>
public class RepairResult
{
public final RepairSession session;
public final RepairJobDesc desc;
public final List<SyncStat> stats;
public RepairFuture(RepairSession session)
public RepairResult(RepairJobDesc desc, List<SyncStat> stats)
{
super(session, null);
this.session = session;
this.desc = desc;
this.stats = stats;
}
}

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

@ -21,13 +21,12 @@ import java.io.IOException;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -35,28 +34,30 @@ import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RepairException;
import org.apache.cassandra.gms.*;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.Pair;
/**
* Coordinates the (active) repair of a token range.
*
* A given RepairSession repairs a set of replicas for a given range on a list
* of column families. For each of the column family to repair, RepairSession
* creates a RepairJob that handles the repair of that CF.
* creates a {@link RepairJob} that handles the repair of that CF.
*
* A given RepairJob has the 2 main phases:
* 1. Validation phase: the job requests merkle trees from each of the replica involves
* (RepairJob.sendTreeRequests()) and waits until all trees are received (in
* <ol>
* <li>Validation phase: the job requests merkle trees from each of the replica involves
* ({@link org.apache.cassandra.repair.ValidationTask}) and waits until all trees are received (in
* validationComplete()).
* 2. Synchonization phase: once all trees are received, the job compares each tree with
* all the other using a so-called Differencer (started by submitDifferencers()). If
* differences there is between 2 trees, the concerned Differencer will start a streaming
* of the difference between the 2 endpoint concerned (Differencer.performStreamingRepair).
* The job is done once all its Differencer are done (i.e. have either computed no differences
* </li>
* <li>Synchronization phase: once all trees are received, the job compares each tree with
* all the other using a so-called {@link SyncTask}. If there is difference between 2 trees, the
* concerned SyncTask will start a streaming of the difference between the 2 endpoint concerned.
* </li>
* </ol>
* The job is done once all its SyncTasks are done (i.e. have either computed no differences
* or the streaming they started is done (syncComplete())).
*
* A given session will execute the first phase (validation phase) of each of it's job
@ -71,15 +72,15 @@ import org.apache.cassandra.utils.concurrent.SimpleCondition;
* we still first send a message to each node to flush and snapshot data so each merkle tree
* creation is still done on similar data, even if the actual creation is not
* done simulatneously). If not sequential, all merkle tree are requested in parallel.
* Similarly, if a job is sequential, it will handle one Differencer at a time, but will handle
* Similarly, if a job is sequential, it will handle one SyncTask at a time, but will handle
* all of them in parallel otherwise.
*/
public class RepairSession extends WrappedRunnable implements IEndpointStateChangeSubscriber,
IFailureDetectionEventListener,
IRepairJobEventListener
public class RepairSession extends AbstractFuture<List<RepairResult>> implements IEndpointStateChangeSubscriber,
IFailureDetectionEventListener
{
private static Logger logger = LoggerFactory.getLogger(RepairSession.class);
public final UUID parentRepairSession;
/** Repair session ID */
private final UUID id;
public final String keyspace;
@ -88,25 +89,18 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan
/** Range to repair */
public final Range<Token> range;
public final Set<InetAddress> endpoints;
private final long repairedAt;
private volatile Exception exception;
private final AtomicBoolean isFailed = new AtomicBoolean(false);
private final AtomicBoolean fdUnregistered = new AtomicBoolean(false);
// First, all RepairJobs are added to this queue,
final Queue<RepairJob> jobs = new ConcurrentLinkedQueue<>();
// and after receiving all validation, the job is moved to
// this map, keyed by CF name.
final Map<String, RepairJob> syncingJobs = new ConcurrentHashMap<>();
// Each validation task waits response from replica in validating ConcurrentMap (keyed by CF name and endpoint address)
private final ConcurrentMap<Pair<RepairJobDesc, InetAddress>, ValidationTask> validating = new ConcurrentHashMap<>();
// Remote syncing jobs wait response in syncingTasks map
private final ConcurrentMap<Pair<RepairJobDesc, NodePair>, RemoteSyncTask> syncingTasks = new ConcurrentHashMap<>();
// Tasks(snapshot, validate request, differencing, ...) are run on taskExecutor
private final ListeningExecutorService taskExecutor = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool(new NamedThreadFactory("RepairJobTask")));
private final SimpleCondition completed = new SimpleCondition();
public final Condition differencingDone = new SimpleCondition();
public final UUID parentRepairSession;
private volatile boolean terminated = false;
/**
@ -118,21 +112,25 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan
* @param endpoints the data centers that should be part of the repair; null for all DCs
* @param cfnames names of columnfamilies
*/
public RepairSession(UUID parentRepairSession, Range<Token> range, String keyspace, boolean isSequential, Set<InetAddress> endpoints, String... cfnames)
public RepairSession(UUID parentRepairSession,
UUID id,
Range<Token> range,
String keyspace,
boolean isSequential,
Set<InetAddress> endpoints,
long repairedAt,
String... cfnames)
{
this(parentRepairSession, UUIDGen.getTimeUUID(), range, keyspace, isSequential, endpoints, cfnames);
}
assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it";
public RepairSession(UUID parentRepairSession, UUID id, Range<Token> range, String keyspace, boolean isSequential, Set<InetAddress> endpoints, String[] cfnames)
{
this.parentRepairSession = parentRepairSession;
this.id = id;
this.isSequential = isSequential;
this.keyspace = keyspace;
this.cfnames = cfnames;
assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it";
this.range = range;
this.endpoints = endpoints;
this.repairedAt = repairedAt;
}
public UUID getId()
@ -145,6 +143,16 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan
return range;
}
public void waitForValidation(Pair<RepairJobDesc, InetAddress> key, ValidationTask task)
{
validating.put(key, task);
}
public void waitForSync(Pair<RepairJobDesc, NodePair> key, RemoteSyncTask task)
{
syncingTasks.put(key, task);
}
/**
* Receive merkle tree response or failed response from {@code endpoint} for current repair job.
*
@ -154,52 +162,15 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan
*/
public void validationComplete(RepairJobDesc desc, InetAddress endpoint, MerkleTree tree)
{
RepairJob job = jobs.peek();
if (job == null)
ValidationTask task = validating.remove(Pair.create(desc, endpoint));
if (task == null)
{
assert terminated;
return;
}
if (tree == null)
{
exception = new RepairException(desc, "Validation failed in " + endpoint);
forceShutdown();
return;
}
logger.info(String.format("[repair #%s] Received merkle tree for %s from %s", getId(), desc.columnFamily, endpoint));
assert job.desc.equals(desc);
if (job.addTree(endpoint, tree) == 0)
{
logger.debug("All responses received for {}/{}", getId(), desc.columnFamily);
if (!job.isFailed())
{
syncingJobs.put(job.desc.columnFamily, job);
job.submitDifferencers();
}
// This job is complete, switching to next in line (note that only one thread will ever do this)
jobs.poll();
RepairJob nextJob = jobs.peek();
if (nextJob == null)
{
// Unregister from FailureDetector once we've completed synchronizing Merkle trees.
// After this point, we rely on tcp_keepalive for individual sockets to notify us when a connection is down.
// See CASSANDRA-3569
if (fdUnregistered.compareAndSet(false, true))
FailureDetector.instance.unregisterFailureDetectionEventListener(this);
// We are done with this repair session as far as differencing
// is considered. Just inform the session
differencingDone.signalAll();
}
else
{
nextJob.sendTreeRequests(endpoints);
}
}
task.treeReceived(tree);
}
/**
@ -211,38 +182,15 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan
*/
public void syncComplete(RepairJobDesc desc, NodePair nodes, boolean success)
{
RepairJob job = syncingJobs.get(desc.columnFamily);
if (job == null)
RemoteSyncTask task = syncingTasks.get(Pair.create(desc, nodes));
if (task == null)
{
assert terminated;
return;
}
if (!success)
{
exception = new RepairException(desc, String.format("Sync failed between %s and %s", nodes.endpoint1, nodes.endpoint2));
forceShutdown();
return;
}
logger.debug(String.format("[repair #%s] Repair completed between %s and %s on %s", getId(), nodes.endpoint1, nodes.endpoint2, desc.columnFamily));
if (job.completedSynchronization())
{
RepairJob completedJob = syncingJobs.remove(job.desc.columnFamily);
String remaining = syncingJobs.size() == 0 ? "" : String.format(" (%d remaining table to sync for this session)", syncingJobs.size());
if (completedJob != null && completedJob.isFailed())
logger.warn(String.format("[repair #%s] %s sync failed%s", getId(), desc.columnFamily, remaining));
else
logger.info(String.format("[repair #%s] %s is fully synced%s", getId(), desc.columnFamily, remaining));
if (jobs.isEmpty() && syncingJobs.isEmpty())
{
taskExecutor.shutdown();
// this repair session is completed
completed.signalAll();
}
}
task.syncComplete(success);
}
private String repairedNodes()
@ -254,15 +202,25 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan
return sb.toString();
}
// we don't care about the return value but care about it throwing exception
public void runMayThrow() throws Exception
/**
* Start RepairJob on given ColumnFamilies.
*
* This first validates if all replica are available, and if they are,
* creates RepairJobs and submit to run on given executor.
*
* @param executor Executor to run validation
*/
public void start(ListeningExecutorService executor)
{
if (terminated)
return;
logger.info(String.format("[repair #%s] new session: will sync %s on range %s for %s.%s", getId(), repairedNodes(), range, keyspace, Arrays.toString(cfnames)));
if (endpoints.isEmpty())
{
differencingDone.signalAll();
logger.info(String.format("[repair #%s] No neighbors to repair with on range %s: session completed", getId(), range));
set(Lists.<RepairResult>newArrayList());
return;
}
@ -272,85 +230,59 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan
if (!FailureDetector.instance.isAlive(endpoint))
{
String message = String.format("Cannot proceed on repair because a neighbor (%s) is dead: session failed", endpoint);
differencingDone.signalAll();
logger.error("[repair #{}] {}", getId(), message);
throw new IOException(message);
setException(new IOException(message));
return;
}
}
ActiveRepairService.instance.addToActiveSessions(this);
try
// Create and submit RepairJob for each ColumnFamily
List<ListenableFuture<RepairResult>> jobs = new ArrayList<>(cfnames.length);
for (String cfname : cfnames)
{
// Create and queue a RepairJob for each column family
for (String cfname : cfnames)
{
RepairJob job = new RepairJob(this, parentRepairSession, id, keyspace, cfname, range, isSequential, taskExecutor);
jobs.offer(job);
}
logger.debug("Sending tree requests to endpoints {}", endpoints);
jobs.peek().sendTreeRequests(endpoints);
RepairJob job = new RepairJob(this, cfname, isSequential, repairedAt, taskExecutor);
executor.execute(job);
jobs.add(job);
}
// block whatever thread started this session until all requests have been returned:
// if this thread dies, the session will still complete in the background
completed.await();
if (exception == null)
// When all RepairJobs are done without error, cleanup and set the final result
Futures.addCallback(Futures.allAsList(jobs), new FutureCallback<List<RepairResult>>()
{
public void onSuccess(List<RepairResult> results)
{
// this repair session is completed
logger.info(String.format("[repair #%s] session completed successfully", getId()));
set(results);
taskExecutor.shutdown();
// mark this session as terminated
terminate();
}
else
public void onFailure(Throwable t)
{
logger.error(String.format("[repair #%s] session completed with the following error", getId()), exception);
throw exception;
logger.error("Repair job failed", t);
setException(t);
}
}
catch (InterruptedException e)
{
throw new RuntimeException("Interrupted while waiting for repair.");
}
finally
{
// mark this session as terminated
terminate();
ActiveRepairService.instance.removeFromActiveSessions(this);
// If we've reached here in an exception state without completing Merkle Tree sync, we'll still be registered
// with the FailureDetector.
if (fdUnregistered.compareAndSet(false, true))
FailureDetector.instance.unregisterFailureDetectionEventListener(this);
}
});
}
public void terminate()
{
terminated = true;
jobs.clear();
syncingJobs.clear();
validating.clear();
syncingTasks.clear();
}
/**
* clear all RepairJobs and terminate this session.
*
* @param reason Cause of error for shutdown
*/
public void forceShutdown()
public void forceShutdown(Throwable reason)
{
setException(reason);
taskExecutor.shutdownNow();
differencingDone.signalAll();
completed.signalAll();
}
public void failedSnapshot()
{
exception = new IOException("Failed during snapshot creation.");
forceShutdown();
}
void failedNode(InetAddress remote)
{
String errorMsg = String.format("Endpoint %s died", remote);
exception = new IOException(errorMsg);
// If a node failed during Merkle creation, we stop everything (though there could still be some activity in the background)
forceShutdown();
terminate();
}
public void onJoin(InetAddress endpoint, EndpointState epState) {}
@ -383,6 +315,9 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan
if (!isFailed.compareAndSet(false, true))
return;
failedNode(endpoint);
Exception exception = new IOException(String.format("Endpoint %s died", endpoint));
logger.error(String.format("[repair #%s] session completed with the following error", getId()), exception);
// If a node failed, we stop everything (though there could still be some activity in the background)
forceShutdown(exception);
}
}

128
src/java/org/apache/cassandra/repair/RequestCoordinator.java
View File

@ -1,128 +0,0 @@
/*
* 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.HashSet;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Set;
/**
*/
public abstract class RequestCoordinator<R>
{
private final Order<R> orderer;
public RequestCoordinator(boolean isSequential)
{
this.orderer = isSequential ? new SequentialOrder(this) : new ParallelOrder(this);
}
public abstract void send(R request);
public void add(R request)
{
orderer.add(request);
}
public void start()
{
orderer.start();
}
// Returns how many request remains
public int completed(R request)
{
return orderer.completed(request);
}
private static abstract class Order<R>
{
protected final RequestCoordinator<R> coordinator;
Order(RequestCoordinator<R> coordinator)
{
this.coordinator = coordinator;
}
public abstract void add(R request);
public abstract void start();
public abstract int completed(R request);
}
private static class SequentialOrder<R> extends Order<R>
{
private final Queue<R> requests = new LinkedList<>();
SequentialOrder(RequestCoordinator<R> coordinator)
{
super(coordinator);
}
public void add(R request)
{
requests.add(request);
}
public void start()
{
if (requests.isEmpty())
return;
coordinator.send(requests.peek());
}
public int completed(R request)
{
assert request.equals(requests.peek());
requests.poll();
int remaining = requests.size();
if (remaining != 0)
coordinator.send(requests.peek());
return remaining;
}
}
private static class ParallelOrder<R> extends Order<R>
{
private final Set<R> requests = new HashSet<>();
ParallelOrder(RequestCoordinator<R> coordinator)
{
super(coordinator);
}
public void add(R request)
{
requests.add(request);
}
public void start()
{
for (R request : requests)
coordinator.send(request);
}
public int completed(R request)
{
requests.remove(request);
return requests.size();
}
}
}

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

@ -74,6 +74,7 @@ public class SnapshotTask extends AbstractFuture<InetAddress> implements Runnabl
public void onFailure(InetAddress from)
{
//listener.failedSnapshot();
task.setException(new RuntimeException("Could not create snapshot at " + from));
}
}

59
src/java/org/apache/cassandra/repair/StreamingRepairTask.java
View File

@ -23,59 +23,40 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.SyncComplete;
import org.apache.cassandra.repair.messages.SyncRequest;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.streaming.StreamEvent;
import org.apache.cassandra.streaming.StreamEventHandler;
import org.apache.cassandra.streaming.StreamPlan;
import org.apache.cassandra.streaming.StreamState;
/**
* Task that make two nodes exchange (stream) some ranges (for a given table/cf).
* This handle the case where the local node is neither of the two nodes that
* must stream their range, and allow to register a callback to be called on
* completion.
* StreamingRepairTask performs data streaming between two remote replica which neither is not repair coordinator.
* Task will send {@link SyncComplete} message back to coordinator upon streaming completion.
*/
public class StreamingRepairTask implements Runnable, StreamEventHandler
{
private static final Logger logger = LoggerFactory.getLogger(StreamingRepairTask.class);
/** Repair session ID that this streaming task belongs */
public final RepairJobDesc desc;
public final SyncRequest request;
private final RepairJobDesc desc;
private final SyncRequest request;
private final long repairedAt;
public StreamingRepairTask(RepairJobDesc desc, SyncRequest request)
public StreamingRepairTask(RepairJobDesc desc, SyncRequest request, long repairedAt)
{
this.desc = desc;
this.request = request;
this.repairedAt = repairedAt;
}
public void run()
{
if (request.src.equals(FBUtilities.getBroadcastAddress()))
initiateStreaming();
else
forwardToSource();
}
private void initiateStreaming()
{
long repairedAt = ActiveRepairService.UNREPAIRED_SSTABLE;
if (desc.parentSessionId != null && ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId) != null)
repairedAt = ActiveRepairService.instance.getParentRepairSession(desc.parentSessionId).repairedAt;
logger.info(String.format("[streaming task #%s] Performing streaming repair of %d ranges with %s", desc.sessionId, request.ranges.size(), request.dst));
StreamResultFuture op = new StreamPlan("Repair", repairedAt, 1)
.flushBeforeTransfer(true)
// request ranges from the remote node
.requestRanges(request.dst, desc.keyspace, request.ranges, desc.columnFamily)
// send ranges to the remote node
.transferRanges(request.dst, desc.keyspace, request.ranges, desc.columnFamily)
.execute();
op.addEventListener(this);
}
private void forwardToSource()
{
logger.info(String.format("[repair #%s] Forwarding streaming repair of %d ranges to %s (to be streamed with %s)", desc.sessionId, request.ranges.size(), request.src, request.dst));
MessagingService.instance().sendOneWay(request.createMessage(), request.src);
new StreamPlan("Repair", repairedAt, 1).listeners(this)
.flushBeforeTransfer(true)
// request ranges from the remote node
.requestRanges(request.dst, desc.keyspace, request.ranges, desc.columnFamily)
// send ranges to the remote node
.transferRanges(request.dst, desc.keyspace, request.ranges, desc.columnFamily)
.execute();
}
public void handleStreamEvent(StreamEvent event)
@ -85,7 +66,7 @@ public class StreamingRepairTask implements Runnable, StreamEventHandler
}
/**
* If we succeeded on both stream in and out, reply back to the initiator.
* If we succeeded on both stream in and out, reply back to coordinator
*/
public void onSuccess(StreamState state)
{
@ -94,7 +75,7 @@ public class StreamingRepairTask implements Runnable, StreamEventHandler
}
/**
* If we failed on either stream in or out, reply fail to the initiator.
* If we failed on either stream in or out, reply fail to coordinator
*/
public void onFailure(Throwable t)
{

18
src/java/org/apache/cassandra/repair/IRepairJobEventListener.java → src/java/org/apache/cassandra/repair/SyncStat.java
View File

@ -18,14 +18,16 @@
package org.apache.cassandra.repair;
/**
* Implemented by the RepairSession to accept callbacks from sequential snapshot creation failure.
* Statistics about synchronizing two replica
*/
public interface IRepairJobEventListener
public class SyncStat
{
/**
* Signal that there was a failure during the snapshot creation process.
*
*/
public void failedSnapshot();
public final NodePair nodes;
public final long numberOfDifferences;
public SyncStat(NodePair nodes, long numberOfDifferences)
{
this.nodes = nodes;
this.numberOfDifferences = numberOfDifferences;
}
}

83
src/java/org/apache/cassandra/repair/SyncTask.java Normal file
View File

@ -0,0 +1,83 @@
/*
* 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.ArrayList;
import java.util.List;
import com.google.common.util.concurrent.AbstractFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.utils.MerkleTree;
/**
* SyncTask will calculate the difference of MerkleTree between two nodes
* and perform necessary operation to repair replica.
*/
public abstract class SyncTask extends AbstractFuture<SyncStat> implements Runnable
{
private static Logger logger = LoggerFactory.getLogger(SyncTask.class);
protected final RepairJobDesc desc;
protected final TreeResponse r1;
protected final TreeResponse r2;
protected volatile SyncStat stat;
public SyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2)
{
this.desc = desc;
this.r1 = r1;
this.r2 = r2;
}
/**
* Compares trees, and triggers repairs for any ranges that mismatch.
*/
public void run()
{
// compare trees, and collect differences
List<Range<Token>> differences = new ArrayList<>();
differences.addAll(MerkleTree.difference(r1.tree, r2.tree));
stat = new SyncStat(new NodePair(r1.endpoint, r2.endpoint), differences.size());
// choose a repair method based on the significance of the difference
String format = String.format("[repair #%s] Endpoints %s and %s %%s for %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily);
if (differences.isEmpty())
{
logger.info(String.format(format, "are consistent"));
set(stat);
return;
}
// non-0 difference: perform streaming repair
logger.info(String.format(format, "have " + differences.size() + " range(s) out of sync"));
startSync(differences);
}
public SyncStat getCurrentStat()
{
return stat;
}
protected abstract void startSync(List<Range<Token>> differences);
}

71
src/java/org/apache/cassandra/repair/ValidationTask.java Normal file
View File

@ -0,0 +1,71 @@
/*
* 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.net.InetAddress;
import com.google.common.util.concurrent.AbstractFuture;
import org.apache.cassandra.exceptions.RepairException;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.ValidationRequest;
import org.apache.cassandra.utils.MerkleTree;
/**
* ValidationTask sends {@link ValidationRequest} to a replica.
* When a replica sends back message, task completes.
*/
public class ValidationTask extends AbstractFuture<TreeResponse> implements Runnable
{
private final RepairJobDesc desc;
private final InetAddress endpoint;
private final int gcBefore;
public ValidationTask(RepairJobDesc desc, InetAddress endpoint, int gcBefore)
{
this.desc = desc;
this.endpoint = endpoint;
this.gcBefore = gcBefore;
}
/**
* Send ValidationRequest to replica
*/
public void run()
{
ValidationRequest request = new ValidationRequest(desc, gcBefore);
MessagingService.instance().sendOneWay(request.createMessage(), endpoint);
}
/**
* Receive MerkleTree from replica node.
*
* @param tree MerkleTree that is sent from replica. Null if validation failed on replica node.
*/
public void treeReceived(MerkleTree tree)
{
if (tree == null)
{
setException(new RepairException(desc, "Validation failed in " + endpoint));
}
else
{
set(new TreeResponse(endpoint, tree));
}
}
}

290
src/java/org/apache/cassandra/repair/messages/RepairOption.java Normal file
View File

@ -0,0 +1,290 @@
/*
* 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.messages;
import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.utils.FBUtilities;
/**
* Repair options.
*/
public class RepairOption
{
public static final String SEQUENTIAL_KEY = "sequential";
public static final String PRIMARY_RANGE_KEY = "primaryRange";
public static final String INCREMENTAL_KEY = "incremental";
public static final String JOB_THREADS_KEY = "jobThreads";
public static final String RANGES_KEY = "ranges";
public static final String COLUMNFAMILIES_KEY = "columnFamilies";
public static final String DATACENTERS_KEY = "dataCenters";
public static final String HOSTS_KEY = "hosts";
// we don't want to push nodes too much for repair
public static final int MAX_JOB_THREADS = 4;
private static final Logger logger = LoggerFactory.getLogger(RepairOption.class);
/**
* Construct RepairOptions object from given map of Strings.
* <p>
* Available options are:
*
* <table>
* <thead>
* <tr>
* <th>key</th>
* <th>value</th>
* <th>default (when key not given)</th>
* </tr>
* </thead>
* <tbody>
* <tr>
* <td>sequential</td>
* <td>"true" if perform sequential repair.</td>
* <td>true</td>
* </tr>
* <tr>
* <td>primaryRange</td>
* <td>"true" if perform repair only on primary range.</td>
* <td>false</td>
* </tr>
* <tr>
* <td>incremental</td>
* <td>"true" if perform incremental repair.</td>
* <td>false</td>
* </tr>
* <tr>
* <td>jobThreads</td>
* <td>Number of threads to use to run repair job.</td>
* <td>1</td>
* </tr>
* <tr>
* <td>ranges</td>
* <td>Ranges to repair. A range is expressed as &lt;start token&gt;:&lt;end token&gt;
* and multiple ranges can be given as comma separated ranges(e.g. aaa:bbb,ccc:ddd).</td>
* <td></td>
* </tr>
* <tr>
* <td>columnFamilies</td>
* <td>Specify names of ColumnFamilies to repair.
* Multiple ColumnFamilies can be given as comma separated values(e.g. cf1,cf2,cf3).</td>
* <td></td>
* </tr>
* <tr>
* <td>dataCenters</td>
* <td>Specify names of data centers who participate in this repair.
* Multiple data centers can be given as comma separated values(e.g. dc1,dc2,dc3).</td>
* <td></td>
* </tr>
* <tr>
* <td>hosts</td>
* <td>Specify names of hosts who participate in this repair.
* Multiple hosts can be given as comma separated values(e.g. cass1,cass2).</td>
* <td></td>
* </tr>
* </tbody>
* </table>
*
* @param options options to parse
* @param partitioner partitioner is used to construct token ranges
* @return RepairOptions object
*/
public static RepairOption parse(Map<String, String> options, IPartitioner partitioner)
{
boolean sequential = !options.containsKey(SEQUENTIAL_KEY) || Boolean.parseBoolean(options.get(SEQUENTIAL_KEY));
boolean primaryRange = Boolean.parseBoolean(options.get(PRIMARY_RANGE_KEY));
boolean incremental = Boolean.parseBoolean(options.get(INCREMENTAL_KEY));
int jobThreads = 1;
if (options.containsKey(JOB_THREADS_KEY))
{
try
{
jobThreads = Integer.parseInt(options.get(JOB_THREADS_KEY));
}
catch (NumberFormatException ignore) {}
}
// ranges
String rangesStr = options.get(RANGES_KEY);
Set<Range<Token>> ranges = new HashSet<>();
if (rangesStr != null)
{
StringTokenizer tokenizer = new StringTokenizer(rangesStr, ",");
while (tokenizer.hasMoreTokens())
{
String[] rangeStr = tokenizer.nextToken().split(":", 2);
if (rangeStr.length < 2)
{
continue;
}
Token parsedBeginToken = partitioner.getTokenFactory().fromString(rangeStr[0].trim());
Token parsedEndToken = partitioner.getTokenFactory().fromString(rangeStr[1].trim());
ranges.add(new Range<>(parsedBeginToken, parsedEndToken));
}
}
RepairOption option = new RepairOption(sequential, primaryRange, incremental, jobThreads, ranges);
// data centers
String dataCentersStr = options.get(DATACENTERS_KEY);
Collection<String> dataCenters = new HashSet<>();
if (dataCentersStr != null)
{
StringTokenizer tokenizer = new StringTokenizer(dataCentersStr, ",");
while (tokenizer.hasMoreTokens())
{
dataCenters.add(tokenizer.nextToken().trim());
}
}
option.getDataCenters().addAll(dataCenters);
// hosts
String hostsStr = options.get(HOSTS_KEY);
Collection<String> hosts = new HashSet<>();
if (hostsStr != null)
{
StringTokenizer tokenizer = new StringTokenizer(hostsStr, ",");
while (tokenizer.hasMoreTokens())
{
hosts.add(tokenizer.nextToken().trim());
}
}
option.getHosts().addAll(hosts);
// columnfamilies
String cfStr = options.get(COLUMNFAMILIES_KEY);
Collection<String> columnFamilies = new HashSet<>();
if (cfStr != null)
{
StringTokenizer tokenizer = new StringTokenizer(cfStr, ",");
while (tokenizer.hasMoreTokens())
{
columnFamilies.add(tokenizer.nextToken().trim());
}
}
option.getColumnFamilies().addAll(columnFamilies);
// validate options
if (jobThreads > MAX_JOB_THREADS)
{
throw new IllegalArgumentException("Too many job threads. Max is " + MAX_JOB_THREADS);
}
if (primaryRange && (!dataCenters.isEmpty() || !hosts.isEmpty()))
{
throw new IllegalArgumentException("You need to run primary range repair on all nodes in the cluster.");
}
return option;
}
private final boolean sequential;
private final boolean primaryRange;
private final boolean incremental;
private final int jobThreads;
private final Collection<String> columnFamilies = new HashSet<>();
private final Collection<String> dataCenters = new HashSet<>();
private final Collection<String> hosts = new HashSet<>();
private final Collection<Range<Token>> ranges = new HashSet<>();
public RepairOption(boolean sequential, boolean primaryRange, boolean incremental, int jobThreads, Collection<Range<Token>> ranges)
{
if (sequential && incremental)
{
String message = "It is not possible to mix sequential repair and incremental repairs.";
logger.error(message);
throw new IllegalArgumentException(message);
}
if (!FBUtilities.isUnix() && sequential)
{
logger.warn("Snapshot-based repair is not yet supported on Windows. Reverting to parallel repair.");
this.sequential = false;
}
else
{
this.sequential = sequential;
}
this.primaryRange = primaryRange;
this.incremental = incremental;
this.jobThreads = jobThreads;
this.ranges.addAll(ranges);
}
public boolean isSequential()
{
return sequential;
}
public boolean isPrimaryRange()
{
return primaryRange;
}
public boolean isIncremental()
{
return incremental;
}
public int getJobThreads()
{
return jobThreads;
}
public Collection<String> getColumnFamilies()
{
return columnFamilies;
}
public Collection<Range<Token>> getRanges()
{
return ranges;
}
public Collection<String> getDataCenters()
{
return dataCenters;
}
public Collection<String> getHosts()
{
return hosts;
}
@Override
public String toString()
{
return "repair options (" +
"sequential: " + sequential +
", primary range: " + primaryRange +
", incremental: " + incremental +
", job threads: " + jobThreads +
", ColumnFamilies: " + columnFamilies +
", dataCenters: " + dataCenters +
", hosts: " + hosts +
", # of ranges: " + ranges.size() +
')';
}
}

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

@ -18,6 +18,7 @@
package org.apache.cassandra.service;
import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
@ -26,12 +27,11 @@ import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.JMXConfigurableThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.dht.Bounds;
@ -39,6 +39,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.IFailureDetector;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.locator.TokenMetadata;
@ -46,12 +47,9 @@ import org.apache.cassandra.net.IAsyncCallbackWithFailure;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.*;
import org.apache.cassandra.repair.messages.AnticompactionRequest;
import org.apache.cassandra.repair.messages.PrepareMessage;
import org.apache.cassandra.repair.messages.RepairMessage;
import org.apache.cassandra.repair.messages.SyncComplete;
import org.apache.cassandra.repair.messages.ValidationComplete;
import org.apache.cassandra.repair.RepairJobDesc;
import org.apache.cassandra.repair.RepairSession;
import org.apache.cassandra.repair.messages.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
@ -73,21 +71,10 @@ public class ActiveRepairService
{
private static final Logger logger = LoggerFactory.getLogger(ActiveRepairService.class);
// singleton enforcement
public static final ActiveRepairService instance = new ActiveRepairService();
public static final ActiveRepairService instance = new ActiveRepairService(FailureDetector.instance, Gossiper.instance);
public static final long UNREPAIRED_SSTABLE = 0;
private static final ThreadPoolExecutor executor;
static
{
executor = new JMXConfigurableThreadPoolExecutor(4,
60,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new NamedThreadFactory("AntiEntropySessions"),
"internal");
}
public static enum Status
{
STARTED, SESSION_SUCCESS, SESSION_FAILED, FINISHED
@ -96,17 +83,17 @@ public class ActiveRepairService
/**
* A map of active coordinator session.
*/
private final ConcurrentMap<UUID, RepairSession> sessions;
private final ConcurrentMap<UUID, RepairSession> sessions = new ConcurrentHashMap<>();
private final ConcurrentMap<UUID, ParentRepairSession> parentRepairSessions;
private final ConcurrentMap<UUID, ParentRepairSession> parentRepairSessions = new ConcurrentHashMap<>();
/**
* Protected constructor. Use ActiveRepairService.instance.
*/
protected ActiveRepairService()
private final IFailureDetector failureDetector;
private final Gossiper gossiper;
public ActiveRepairService(IFailureDetector failureDetector, Gossiper gossiper)
{
sessions = new ConcurrentHashMap<>();
parentRepairSessions = new ConcurrentHashMap<>();
this.failureDetector = failureDetector;
this.gossiper = gossiper;
}
/**
@ -114,51 +101,52 @@ public class ActiveRepairService
*
* @return Future for asynchronous call or null if there is no need to repair
*/
public RepairFuture submitRepairSession(UUID parentRepairSession, Range<Token> range, String keyspace, boolean isSequential, Set<InetAddress> endpoints, String... cfnames)
public RepairSession submitRepairSession(UUID parentRepairSession,
Range<Token> range,
String keyspace,
boolean isSequential,
Set<InetAddress> endpoints,
long repairedAt,
ListeningExecutorService executor,
String... cfnames)
{
RepairSession session = new RepairSession(parentRepairSession, range, keyspace, isSequential, endpoints, cfnames);
if (session.endpoints.isEmpty())
if (endpoints.isEmpty())
return null;
RepairFuture futureTask = new RepairFuture(session);
executor.execute(futureTask);
return futureTask;
}
public void addToActiveSessions(RepairSession session)
{
final RepairSession session = new RepairSession(parentRepairSession, UUIDGen.getTimeUUID(), range, keyspace, isSequential, endpoints, repairedAt, cfnames);
sessions.put(session.getId(), session);
Gossiper.instance.register(session);
FailureDetector.instance.registerFailureDetectionEventListener(session);
}
// register listeners
gossiper.register(session);
failureDetector.registerFailureDetectionEventListener(session);
public void removeFromActiveSessions(RepairSession session)
{
Gossiper.instance.unregister(session);
sessions.remove(session.getId());
// unregister listeners at completion
session.addListener(new Runnable()
{
/**
* When repair finished, do clean up
*/
public void run()
{
failureDetector.unregisterFailureDetectionEventListener(session);
gossiper.unregister(session);
sessions.remove(session.getId());
}
}, MoreExecutors.sameThreadExecutor());
session.start(executor);
return session;
}
public void terminateSessions()
{
Throwable cause = new IOException("Terminate session is called");
for (RepairSession session : sessions.values())
{
session.forceShutdown();
session.forceShutdown(cause);
}
parentRepairSessions.clear();
}
// for testing only. Create a session corresponding to a fake request and
// add it to the sessions (avoid NPE in tests)
RepairFuture submitArtificialRepairSession(RepairJobDesc desc)
{
Set<InetAddress> neighbours = new HashSet<>();
neighbours.addAll(ActiveRepairService.getNeighbors(desc.keyspace, desc.range, null, null));
RepairSession session = new RepairSession(desc.parentSessionId, desc.sessionId, desc.range, desc.keyspace, false, neighbours, new String[]{desc.columnFamily});
sessions.put(session.getId(), session);
RepairFuture futureTask = new RepairFuture(session);
executor.execute(futureTask);
return futureTask;
}
/**
* Return all of the neighbors with whom we share the provided range.
*
@ -191,7 +179,7 @@ public class ActiveRepairService
Set<InetAddress> neighbors = new HashSet<>(replicaSets.get(rangeSuperSet));
neighbors.remove(FBUtilities.getBroadcastAddress());
if (dataCenters != null)
if (dataCenters != null && !dataCenters.isEmpty())
{
TokenMetadata.Topology topology = ss.getTokenMetadata().cloneOnlyTokenMap().getTopology();
Set<InetAddress> dcEndpoints = Sets.newHashSet();
@ -204,7 +192,7 @@ public class ActiveRepairService
}
return Sets.intersection(neighbors, dcEndpoints);
}
else if (hosts != null)
else if (hosts != null && !hosts.isEmpty())
{
Set<InetAddress> specifiedHost = new HashSet<>();
for (final String host : hosts)
@ -314,21 +302,18 @@ public class ActiveRepairService
parentRepairSessions.put(parentRepairSession, new ParentRepairSession(columnFamilyStores, ranges, sstablesToRepair, System.currentTimeMillis()));
}
public void finishParentSession(UUID parentSession, Set<InetAddress> neighbors, boolean doAntiCompaction)
public void finishParentSession(UUID parentSession, Set<InetAddress> neighbors)
{
try
{
if (doAntiCompaction)
for (InetAddress neighbor : neighbors)
{
for (InetAddress neighbor : neighbors)
{
AnticompactionRequest acr = new AnticompactionRequest(parentSession);
MessageOut<RepairMessage> req = acr.createMessage();
MessagingService.instance().sendOneWay(req, neighbor);
}
List<Future<?>> futures = doAntiCompaction(parentSession);
FBUtilities.waitOnFutures(futures);
AnticompactionRequest acr = new AnticompactionRequest(parentSession);
MessageOut<RepairMessage> req = acr.createMessage();
MessagingService.instance().sendOneWay(req, neighbor);
}
List<Future<?>> futures = doAntiCompaction(parentSession);
FBUtilities.waitOnFutures(futures);
}
finally
{
@ -407,7 +392,7 @@ public class ActiveRepairService
this.repairedAt = repairedAt;
}
public Collection<SSTableReader> getAndReferenceSSTables(UUID cfId)
public synchronized Collection<SSTableReader> getAndReferenceSSTables(UUID cfId)
{
Set<SSTableReader> sstables = sstableMap.get(cfId);
Iterator<SSTableReader> sstableIterator = sstables.iterator();

330
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -30,6 +30,7 @@ import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;
import javax.management.JMX;
import javax.management.MBeanServer;
import javax.management.Notification;
@ -46,17 +47,13 @@ import ch.qos.logback.core.Appender;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Predicate;
import com.google.common.collect.*;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.Uninterruptibles;
import com.google.common.util.concurrent.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DurationFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.auth.Auth;
import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.concurrent.*;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
@ -80,8 +77,10 @@ import org.apache.cassandra.net.AsyncOneResponse;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.ResponseVerbHandler;
import org.apache.cassandra.repair.RepairFuture;
import org.apache.cassandra.repair.RepairMessageVerbHandler;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.repair.RepairResult;
import org.apache.cassandra.repair.RepairSession;
import org.apache.cassandra.service.paxos.CommitVerbHandler;
import org.apache.cassandra.service.paxos.PrepareVerbHandler;
import org.apache.cassandra.service.paxos.ProposeVerbHandler;
@ -2497,87 +2496,128 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
sendNotification(jmxNotification);
}
public int forceRepairAsync(String keyspace, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean primaryRange, boolean fullRepair, String... columnFamilies) throws IOException
public int repairAsync(String keyspace, Map<String, String> repairSpec)
{
Collection<Range<Token>> ranges;
if (primaryRange)
RepairOption option = RepairOption.parse(repairSpec, getPartitioner());
// if ranges are not specified
if (option.getRanges().isEmpty())
{
// when repairing only primary range, neither dataCenters nor hosts can be set
if (dataCenters == null && hosts == null)
ranges = getPrimaryRanges(keyspace);
// except dataCenters only contain local DC (i.e. -local)
else if (dataCenters != null && dataCenters.size() == 1 && dataCenters.contains(DatabaseDescriptor.getLocalDataCenter()))
ranges = getPrimaryRangesWithinDC(keyspace);
else
throw new IllegalArgumentException("You need to run primary range repair on all nodes in the cluster.");
}
else
{
ranges = getLocalRanges(keyspace);
}
return forceRepairAsync(keyspace, isSequential, dataCenters, hosts, ranges, fullRepair, columnFamilies);
}
public int forceRepairAsync(String keyspace, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, Collection<Range<Token>> ranges, boolean fullRepair, String... columnFamilies)
{
if (ranges.isEmpty() || Keyspace.open(keyspace).getReplicationStrategy().getReplicationFactor() < 2)
return 0;
int cmd = nextRepairCommand.incrementAndGet();
if (ranges.size() > 0)
{
if (!FBUtilities.isUnix() && isSequential)
if (option.isPrimaryRange())
{
logger.warn("Snapshot-based repair is not yet supported on Windows. Reverting to parallel repair.");
isSequential = false;
// when repairing only primary range, neither dataCenters nor hosts can be set
if (option.getDataCenters().isEmpty() && option.getHosts().isEmpty())
option.getRanges().addAll(getPrimaryRanges(keyspace));
// except dataCenters only contain local DC (i.e. -local)
else if (option.getDataCenters().size() == 1 && option.getDataCenters().contains(DatabaseDescriptor.getLocalDataCenter()))
option.getRanges().addAll(getPrimaryRangesWithinDC(keyspace));
else
throw new IllegalArgumentException("You need to run primary range repair on all nodes in the cluster.");
}
else
{
option.getRanges().addAll(getLocalRanges(keyspace));
}
new Thread(createRepairTask(cmd, keyspace, ranges, isSequential, dataCenters, hosts, fullRepair, columnFamilies)).start();
}
return cmd;
return forceRepairAsync(keyspace, option);
}
public int forceRepairAsync(String keyspace, boolean isSequential, boolean isLocal, boolean primaryRange, boolean fullRepair, String... columnFamilies)
public int forceRepairAsync(String keyspace,
boolean isSequential,
Collection<String> dataCenters,
Collection<String> hosts,
boolean primaryRange,
boolean fullRepair,
String... columnFamilies)
{
Collection<Range<Token>> ranges;
if (primaryRange)
if (!FBUtilities.isUnix() && isSequential)
{
ranges = isLocal ? getPrimaryRangesWithinDC(keyspace) : getPrimaryRanges(keyspace);
logger.warn("Snapshot-based repair is not yet supported on Windows. Reverting to parallel repair.");
isSequential = false;
}
else
RepairOption options = new RepairOption(isSequential, primaryRange, !fullRepair, 1, Collections.<Range<Token>>emptyList());
if (dataCenters != null)
{
ranges = getLocalRanges(keyspace);
options.getDataCenters().addAll(dataCenters);
}
return forceRepairAsync(keyspace, isSequential, isLocal, ranges, fullRepair, columnFamilies);
if (hosts != null)
{
options.getHosts().addAll(hosts);
}
if (columnFamilies != null)
{
for (String columnFamily : columnFamilies)
{
options.getColumnFamilies().add(columnFamily);
}
}
return forceRepairAsync(keyspace, options);
}
public int forceRepairAsync(final String keyspace, final boolean isSequential, final boolean isLocal, final Collection<Range<Token>> ranges, final boolean fullRepair, final String... columnFamilies)
public int forceRepairAsync(String keyspace,
boolean isSequential,
boolean isLocal,
boolean primaryRange,
boolean fullRepair,
String... columnFamilies)
{
if (ranges.isEmpty() || Keyspace.open(keyspace).getReplicationStrategy().getReplicationFactor() < 2)
return 0;
int cmd = nextRepairCommand.incrementAndGet();
new Thread(createRepairTask(cmd, keyspace, ranges, isSequential, isLocal, fullRepair, columnFamilies)).start();
return cmd;
Set<String> dataCenters = null;
if (isLocal)
{
dataCenters = Sets.newHashSet(DatabaseDescriptor.getLocalDataCenter());
}
return forceRepairAsync(keyspace, isSequential, dataCenters, null, primaryRange, fullRepair, columnFamilies);
}
public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean fullRepair, String... columnFamilies) throws IOException
public int forceRepairRangeAsync(String beginToken,
String endToken,
String keyspaceName,
boolean isSequential,
Collection<String> dataCenters,
Collection<String> hosts,
boolean fullRepair,
String... columnFamilies)
{
if (!FBUtilities.isUnix() && isSequential)
{
logger.warn("Snapshot-based repair is not yet supported on Windows. Reverting to parallel repair.");
isSequential = false;
}
Collection<Range<Token>> repairingRange = createRepairRangeFrom(beginToken, endToken);
RepairOption options = new RepairOption(isSequential, false, !fullRepair, 1, repairingRange);
options.getDataCenters().addAll(dataCenters);
if (hosts != null)
{
options.getHosts().addAll(hosts);
}
if (columnFamilies != null)
{
for (String columnFamily : columnFamilies)
{
options.getColumnFamilies().add(columnFamily);
}
}
logger.info("starting user-requested repair of range {} for keyspace {} and column families {}",
repairingRange, keyspaceName, columnFamilies);
return forceRepairAsync(keyspaceName, isSequential, dataCenters, hosts, repairingRange, fullRepair, columnFamilies);
return forceRepairAsync(keyspaceName, options);
}
public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, boolean isLocal, boolean fullRepair, String... columnFamilies)
public int forceRepairRangeAsync(String beginToken,
String endToken,
String keyspaceName,
boolean isSequential,
boolean isLocal,
boolean fullRepair,
String... columnFamilies)
{
Collection<Range<Token>> repairingRange = createRepairRangeFrom(beginToken, endToken);
logger.info("starting user-requested repair of range {} for keyspace {} and column families {}",
repairingRange, keyspaceName, columnFamilies);
return forceRepairAsync(keyspaceName, isSequential, isLocal, repairingRange, fullRepair, columnFamilies);
Set<String> dataCenters = null;
if (isLocal)
{
dataCenters = Sets.newHashSet(DatabaseDescriptor.getLocalDataCenter());
}
return forceRepairRangeAsync(beginToken, endToken, keyspaceName, isSequential, dataCenters, null, fullRepair, columnFamilies);
}
/**
@ -2619,32 +2659,19 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return repairingRange;
}
private FutureTask<Object> createRepairTask(int cmd,
String keyspace,
Collection<Range<Token>> ranges,
boolean isSequential,
boolean isLocal,
boolean fullRepair,
String... columnFamilies)
public int forceRepairAsync(String keyspace, RepairOption options)
{
Set<String> dataCenters = null;
if (isLocal)
{
dataCenters = Sets.newHashSet(DatabaseDescriptor.getLocalDataCenter());
}
return createRepairTask(cmd, keyspace, ranges, isSequential, dataCenters, null, fullRepair, columnFamilies);
if (options.getRanges().isEmpty() || Keyspace.open(keyspace).getReplicationStrategy().getReplicationFactor() < 2)
return 0;
int cmd = nextRepairCommand.incrementAndGet();
new Thread(createRepairTask(cmd, keyspace, options)).start();
return cmd;
}
private FutureTask<Object> createRepairTask(final int cmd,
final String keyspace,
final Collection<Range<Token>> ranges,
final boolean isSequential,
final Collection<String> dataCenters,
final Collection<String> hosts,
final boolean fullRepair,
final String... columnFamilies)
private FutureTask<Object> createRepairTask(final int cmd, final String keyspace, final RepairOption options)
{
if (dataCenters != null && !dataCenters.contains(DatabaseDescriptor.getLocalDataCenter()))
if (!options.getDataCenters().isEmpty() && options.getDataCenters().contains(DatabaseDescriptor.getLocalDataCenter()))
{
throw new IllegalArgumentException("the local data center must be part of the repair");
}
@ -2653,11 +2680,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
{
protected void runMayThrow() throws Exception
{
String message = String.format("Starting repair command #%d, repairing %d ranges for keyspace %s (seq=%b, full=%b)", cmd, ranges.size(), keyspace, isSequential, fullRepair);
final long startTime = System.currentTimeMillis();
String message = String.format("Starting repair command #%d, repairing keyspace %s with %s", cmd, keyspace, options);
logger.info(message);
sendNotification("repair", message, new int[]{cmd, ActiveRepairService.Status.STARTED.ordinal()});
if (isSequential && !fullRepair)
if (options.isSequential() && options.isIncremental())
{
message = "It is not possible to mix sequential repair and incremental repairs.";
logger.error(message);
@ -2665,13 +2693,13 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return;
}
Set<InetAddress> allNeighbors = new HashSet<>();
final Set<InetAddress> allNeighbors = new HashSet<>();
Map<Range, Set<InetAddress>> rangeToNeighbors = new HashMap<>();
for (Range<Token> range : ranges)
for (Range<Token> range : options.getRanges())
{
try
{
Set<InetAddress> neighbors = ActiveRepairService.getNeighbors(keyspace, range, dataCenters, hosts);
Set<InetAddress> neighbors = ActiveRepairService.getNeighbors(keyspace, range, options.getDataCenters(), options.getHosts());
rangeToNeighbors.put(range, neighbors);
allNeighbors.addAll(neighbors);
}
@ -2685,6 +2713,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
// Validate columnfamilies
List<ColumnFamilyStore> columnFamilyStores = new ArrayList<>();
String[] columnFamilies = options.getColumnFamilies().toArray(new String[options.getColumnFamilies().size()]);
try
{
Iterables.addAll(columnFamilyStores, getValidColumnFamilies(false, false, keyspace, columnFamilies));
@ -2695,12 +2724,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return;
}
UUID parentSession = null;
if (!fullRepair)
final UUID parentSession;
long repairedAt = ActiveRepairService.UNREPAIRED_SSTABLE;
if (options.isIncremental())
{
try
{
parentSession = ActiveRepairService.instance.prepareForRepair(allNeighbors, ranges, columnFamilyStores);
parentSession = ActiveRepairService.instance.prepareForRepair(allNeighbors, options.getRanges(), columnFamilyStores);
repairedAt = ActiveRepairService.instance.getParentRepairSession(parentSession).repairedAt;
}
catch (Throwable t)
{
@ -2708,60 +2739,93 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return;
}
}
else
{
parentSession = null;
}
List<RepairFuture> futures = new ArrayList<>(ranges.size());
// Set up RepairJob executor for this repair command.
final ListeningExecutorService executor = MoreExecutors.listeningDecorator(new JMXConfigurableThreadPoolExecutor(options.getJobThreads(),
Integer.MAX_VALUE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new NamedThreadFactory("Repair#" + cmd),
"internal"));
List<ListenableFuture<?>> futures = new ArrayList<>(options.getRanges().size());
String[] cfnames = new String[columnFamilyStores.size()];
for (int i = 0; i < columnFamilyStores.size(); i++)
{
cfnames[i] = columnFamilyStores.get(i).name;
}
for (Range<Token> range : ranges)
for (Range<Token> range : options.getRanges())
{
RepairFuture future = ActiveRepairService.instance.submitRepairSession(parentSession, range, keyspace, isSequential, rangeToNeighbors.get(range), cfnames);
if (future == null)
final RepairSession session = ActiveRepairService.instance.submitRepairSession(parentSession,
range,
keyspace,
options.isSequential(),
rangeToNeighbors.get(range),
repairedAt,
executor,
cfnames);
if (session == null)
continue;
futures.add(future);
// wait for a session to be done with its differencing before starting the next one
try
// After repair session completes, notify client its result
Futures.addCallback(session, new FutureCallback<List<RepairResult>>()
{
future.session.differencingDone.await();
}
catch (InterruptedException e)
{
message = "Interrupted while waiting for the differencing of repair session " + future.session + " to be done. Repair may be imprecise.";
logger.error(message, e);
sendNotification("repair", message, new int[]{cmd, ActiveRepairService.Status.SESSION_FAILED.ordinal()});
}
public void onSuccess(List<RepairResult> results)
{
String message = String.format("Repair session %s for range %s finished", session.getId(), session.getRange().toString());
logger.info(message);
sendNotification("repair", message, new int[]{cmd, ActiveRepairService.Status.SESSION_SUCCESS.ordinal()});
}
public void onFailure(Throwable t)
{
String message = String.format("Repair session %s for range %s failed with error %s", session.getId(), session.getRange().toString(), t.getMessage());
logger.error(message, t);
sendNotification("repair", message, new int[]{cmd, ActiveRepairService.Status.SESSION_FAILED.ordinal()});
}
});
futures.add(session);
}
boolean successful = true;
for (RepairFuture future : futures)
// After all repair sessions completes(successful or not),
// run anticompaction if necessary and send finish notice back to client
ListenableFuture<?> allSessions = Futures.allAsList(futures);
Futures.addCallback(allSessions, new FutureCallback<Object>()
{
try
public void onSuccess(@Nullable Object result)
{
future.get();
message = String.format("Repair session %s for range %s finished", future.session.getId(), future.session.getRange().toString());
if (options.isIncremental())
{
try
{
ActiveRepairService.instance.finishParentSession(parentSession, allNeighbors);
}
catch (Exception e)
{
logger.error("Error in incremental repair", e);
}
}
repairComplete();
}
public void onFailure(Throwable t)
{
repairComplete();
}
private void repairComplete()
{
String duration = DurationFormatUtils.formatDurationWords(System.currentTimeMillis() - startTime, true, true);
String message = String.format("Repair command #%d finished in %s", cmd, duration);
sendNotification("repair", message,
new int[]{cmd, ActiveRepairService.Status.FINISHED.ordinal()});
logger.info(message);
sendNotification("repair", message, new int[]{cmd, ActiveRepairService.Status.SESSION_SUCCESS.ordinal()});
executor.shutdownNow();
}
catch (ExecutionException e)
{
successful = false;
message = String.format("Repair session %s for range %s failed with error %s", future.session.getId(), future.session.getRange().toString(), e.getCause().getMessage());
logger.error(message, e);
sendNotification("repair", message, new int[]{cmd, ActiveRepairService.Status.SESSION_FAILED.ordinal()});
}
catch (Exception e)
{
successful = false;
message = String.format("Repair session %s for range %s failed with error %s", future.session.getId(), future.session.getRange().toString(), e.getMessage());
logger.error(message, e);
sendNotification("repair", message, new int[]{cmd, ActiveRepairService.Status.SESSION_FAILED.ordinal()});
}
}
if (!fullRepair)
ActiveRepairService.instance.finishParentSession(parentSession, allNeighbors, successful);
sendNotification("repair", String.format("Repair command #%d finished", cmd), new int[]{cmd, ActiveRepairService.Status.FINISHED.ordinal()});
}, MoreExecutors.sameThreadExecutor());
}
}, null);
}

23
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -270,29 +270,22 @@ public interface StorageServiceMBean extends NotificationEmitter
* type: "repair"
* userObject: int array of length 2, [0]=command number, [1]=ordinal of AntiEntropyService.Status
*
* @param keyspace Keyspace name to repair. Should not be null.
* @param options repair option.
* @return Repair command number, or 0 if nothing to repair
*/
public int repairAsync(String keyspace, Map<String, String> options);
@Deprecated
public int forceRepairAsync(String keyspace, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean primaryRange, boolean repairedAt, String... columnFamilies) throws IOException;
/**
* Same as forceRepairAsync, but handles a specified range
*/
@Deprecated
public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean repairedAt, String... columnFamilies) throws IOException;
/**
* Invoke repair asynchronously.
* You can track repair progress by subscribing JMX notification sent from this StorageServiceMBean.
* Notification format is:
* type: "repair"
* userObject: int array of length 2, [0]=command number, [1]=ordinal of AntiEntropyService.Status
*
* @return Repair command number, or 0 if nothing to repair
*/
@Deprecated
public int forceRepairAsync(String keyspace, boolean isSequential, boolean isLocal, boolean primaryRange, boolean fullRepair, String... columnFamilies);
/**
* Same as forceRepairAsync, but handles a specified range
*/
@Deprecated
public int forceRepairRangeAsync(String beginToken, String endToken, String keyspaceName, boolean isSequential, boolean isLocal, boolean repairedAt, String... columnFamilies);
public void forceTerminateAllRepairSessions();

11
src/java/org/apache/cassandra/streaming/StreamReceiveTask.java
View File

@ -21,16 +21,15 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableWriter;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
/**
@ -38,9 +37,7 @@ import org.apache.cassandra.utils.Pair;
*/
public class StreamReceiveTask extends StreamTask
{
private static final ThreadPoolExecutor executor = DebuggableThreadPoolExecutor.createWithMaximumPoolSize("StreamReceiveTask",
FBUtilities.getAvailableProcessors(),
60, TimeUnit.SECONDS);
private static final ExecutorService executor = Executors.newCachedThreadPool(new NamedThreadFactory("StreamReceiveTask"));
// number of files to receive
private final int totalFiles;

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

@ -43,6 +43,7 @@ import com.google.common.base.Function;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.AbstractFuture;
import com.yammer.metrics.reporting.JmxReporter;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutorMBean;
import org.apache.cassandra.db.ColumnFamilyStoreMBean;
@ -249,43 +250,15 @@ public class NodeProbe implements AutoCloseable
ssProxy.forceKeyspaceFlush(keyspaceName, columnFamilies);
}
public void forceRepairAsync(final PrintStream out, final String keyspaceName, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean primaryRange, boolean fullRepair, String... columnFamilies) throws IOException
public void repairAsync(final PrintStream out, final String keyspace, Map<String, String> options) throws IOException
{
RepairRunner runner = new RepairRunner(out, keyspaceName, columnFamilies);
RepairRunner runner = new RepairRunner(out, ssProxy, keyspace, options);
try
{
jmxc.addConnectionNotificationListener(runner, null, null);
ssProxy.addNotificationListener(runner, null, null);
if (!runner.repairAndWait(ssProxy, isSequential, dataCenters, hosts, primaryRange, fullRepair))
failed = true;
}
catch (Exception e)
{
throw new IOException(e) ;
}
finally
{
try
{
ssProxy.removeNotificationListener(runner);
jmxc.removeConnectionNotificationListener(runner);
}
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(t);
out.println("Exception occurred during clean-up. " + t);
}
}
}
public void forceRepairRangeAsync(final PrintStream out, final String keyspaceName, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, final String startToken, final String endToken, boolean fullRepair, String... columnFamilies) throws IOException
{
RepairRunner runner = new RepairRunner(out, keyspaceName, columnFamilies);
try
{
jmxc.addConnectionNotificationListener(runner, null, null);
ssProxy.addNotificationListener(runner, null, null);
if (!runner.repairRangeAndWait(ssProxy, isSequential, dataCenters, hosts, startToken, endToken, fullRepair))
runner.run();
if (!runner.get())
failed = true;
}
catch (Exception e)
@ -1273,88 +1246,3 @@ class ThreadPoolProxyMBeanIterator implements Iterator<Map.Entry<String, JMXEnab
throw new UnsupportedOperationException();
}
}
class RepairRunner implements NotificationListener
{
private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss,SSS");
private final Condition condition = new SimpleCondition();
private final PrintStream out;
private final String keyspace;
private final String[] columnFamilies;
private int cmd;
private volatile boolean success = true;
private volatile Exception error = null;
RepairRunner(PrintStream out, String keyspace, String... columnFamilies)
{
this.out = out;
this.keyspace = keyspace;
this.columnFamilies = columnFamilies;
}
public boolean repairAndWait(StorageServiceMBean ssProxy, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, boolean primaryRangeOnly, boolean fullRepair) throws Exception
{
cmd = ssProxy.forceRepairAsync(keyspace, isSequential, dataCenters, hosts, primaryRangeOnly, fullRepair, columnFamilies);
waitForRepair();
return success;
}
public boolean repairRangeAndWait(StorageServiceMBean ssProxy, boolean isSequential, Collection<String> dataCenters, Collection<String> hosts, String startToken, String endToken, boolean fullRepair) throws Exception
{
cmd = ssProxy.forceRepairRangeAsync(startToken, endToken, keyspace, isSequential, dataCenters, hosts, fullRepair, columnFamilies);
waitForRepair();
return success;
}
private void waitForRepair() throws Exception
{
if (cmd > 0)
{
condition.await();
}
else
{
String message = String.format("[%s] Nothing to repair for keyspace '%s'", format.format(System.currentTimeMillis()), keyspace);
out.println(message);
}
if (error != null)
{
throw error;
}
}
public void handleNotification(Notification notification, Object handback)
{
if ("repair".equals(notification.getType()))
{
int[] status = (int[]) notification.getUserData();
assert status.length == 2;
if (cmd == status[0])
{
String message = String.format("[%s] %s", format.format(notification.getTimeStamp()), notification.getMessage());
out.println(message);
// repair status is int array with [0] = cmd number, [1] = status
if (status[1] == ActiveRepairService.Status.SESSION_FAILED.ordinal())
success = false;
else if (status[1] == ActiveRepairService.Status.FINISHED.ordinal())
condition.signalAll();
}
}
else if (JMXConnectionNotification.NOTIFS_LOST.equals(notification.getType()))
{
String message = String.format("[%s] Lost notification. You should check server log for repair status of keyspace %s",
format.format(notification.getTimeStamp()),
keyspace);
out.println(message);
}
else if (JMXConnectionNotification.FAILED.equals(notification.getType())
|| JMXConnectionNotification.CLOSED.equals(notification.getType()))
{
String message = String.format("JMX connection closed. You should check server log for repair status of keyspace %s"
+ "(Subsequent keyspaces are not going to be repaired).",
keyspace);
error = new IOException(message);
condition.signalAll();
}
}
}

39
src/java/org/apache/cassandra/tools/NodeTool.java
View File

@ -37,6 +37,7 @@ import com.google.common.collect.Maps;
import com.yammer.metrics.reporting.JmxReporter;
import io.airlift.command.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutorMBean;
import org.apache.cassandra.config.Schema;
@ -48,6 +49,7 @@ import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.EndpointSnitchInfoMBean;
import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.net.MessagingServiceMBean;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.service.CacheServiceMBean;
import org.apache.cassandra.streaming.ProgressInfo;
import org.apache.cassandra.streaming.SessionInfo;
@ -1677,6 +1679,11 @@ public class NodeTool
@Option(title = "full", name = {"-full", "--full"}, description = "Use -full to issue a full repair.")
private boolean fullRepair = false;
@Option(title = "job_threads", name = {"-j", "--job-threads"}, description = "Number of threads to run repair jobs. " +
"Usually this means number of CFs to repair concurrently. " +
"WARNING: increasing this puts more load on repairing nodes, so be careful. (default: 1, max: 4)")
private int numJobThreads = 1;
@Override
public void execute(NodeProbe probe)
{
@ -1688,20 +1695,28 @@ public class NodeTool
for (String keyspace : keyspaces)
{
Map<String, String> options = new HashMap<>();
options.put(RepairOption.SEQUENTIAL_KEY, Boolean.toString(sequential));
options.put(RepairOption.PRIMARY_RANGE_KEY, Boolean.toString(primaryRange));
options.put(RepairOption.INCREMENTAL_KEY, Boolean.toString(!fullRepair));
options.put(RepairOption.JOB_THREADS_KEY, Integer.toString(numJobThreads));
options.put(RepairOption.COLUMNFAMILIES_KEY, StringUtils.join(cfnames, ","));
if (!startToken.isEmpty() || !endToken.isEmpty())
{
options.put(RepairOption.RANGES_KEY, startToken + ":" + endToken);
}
if (localDC)
{
options.put(RepairOption.DATACENTERS_KEY, StringUtils.join(newArrayList(probe.getDataCenter()), ","));
}
else
{
options.put(RepairOption.DATACENTERS_KEY, StringUtils.join(specificDataCenters, ","));
}
options.put(RepairOption.HOSTS_KEY, StringUtils.join(specificHosts, ","));
try
{
Collection<String> dataCenters = null;
Collection<String> hosts = null;
if (!specificDataCenters.isEmpty())
dataCenters = newArrayList(specificDataCenters);
else if (localDC)
dataCenters = newArrayList(probe.getDataCenter());
else if(!specificHosts.isEmpty())
hosts = newArrayList(specificHosts);
if (!startToken.isEmpty() || !endToken.isEmpty())
probe.forceRepairRangeAsync(System.out, keyspace, sequential, dataCenters,hosts, startToken, endToken, fullRepair);
else
probe.forceRepairAsync(System.out, keyspace, sequential, dataCenters, hosts, primaryRange, fullRepair, cfnames);
probe.repairAsync(System.out, keyspace, options);
} catch (Exception e)
{
throw new RuntimeException("Error occurred during repair", e);

101
src/java/org/apache/cassandra/tools/RepairRunner.java Normal file
View File

@ -0,0 +1,101 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.tools;
import java.io.IOException;
import java.io.PrintStream;
import java.text.SimpleDateFormat;
import java.util.Map;
import javax.management.Notification;
import javax.management.NotificationListener;
import javax.management.remote.JMXConnectionNotification;
import com.google.common.util.concurrent.AbstractFuture;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageServiceMBean;
public class RepairRunner extends AbstractFuture<Boolean> implements Runnable, NotificationListener
{
private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss,SSS");
private final PrintStream out;
private final StorageServiceMBean ssProxy;
private final String keyspace;
private final Map<String, String> options;
private volatile int cmd;
private volatile boolean success;
public RepairRunner(PrintStream out, StorageServiceMBean ssProxy, String keyspace, Map<String, String> options)
{
this.out = out;
this.ssProxy = ssProxy;
this.keyspace = keyspace;
this.options = options;
}
public void run()
{
cmd = ssProxy.repairAsync(keyspace, options);
if (cmd <= 0)
{
String message = String.format("[%s] Nothing to repair for keyspace '%s'", format.format(System.currentTimeMillis()), keyspace);
out.println(message);
set(true);
}
}
public void handleNotification(Notification notification, Object handback)
{
if ("repair".equals(notification.getType()))
{
int[] status = (int[]) notification.getUserData();
assert status.length == 2;
if (cmd == status[0])
{
String message = String.format("[%s] %s", format.format(notification.getTimeStamp()), notification.getMessage());
out.println(message);
// repair status is int array with [0] = cmd number, [1] = status
if (status[1] == ActiveRepairService.Status.SESSION_FAILED.ordinal())
{
success = false;
}
else if (status[1] == ActiveRepairService.Status.FINISHED.ordinal())
{
set(success);
}
}
}
else if (JMXConnectionNotification.NOTIFS_LOST.equals(notification.getType()))
{
String message = String.format("[%s] Lost notification. You should check server log for repair status of keyspace %s",
format.format(notification.getTimeStamp()),
keyspace);
out.println(message);
}
else if (JMXConnectionNotification.FAILED.equals(notification.getType())
|| JMXConnectionNotification.CLOSED.equals(notification.getType()))
{
String message = String.format("JMX connection closed. You should check server log for repair status of keyspace %s"
+ "(Subsequent keyspaces are not going to be repaired).",
keyspace);
setException(new IOException(message));
}
}
}

51
test/unit/org/apache/cassandra/repair/DifferencerTest.java → test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java
View File

@ -23,7 +23,6 @@ import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import org.junit.After;
import org.junit.BeforeClass;
import org.junit.Test;
@ -36,20 +35,12 @@ import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.SimpleStrategy;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.sink.IMessageSink;
import org.apache.cassandra.sink.SinkManager;
import org.apache.cassandra.repair.messages.RepairMessage;
import org.apache.cassandra.repair.messages.SyncComplete;
import org.apache.cassandra.utils.MerkleTree;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class DifferencerTest
public class LocalSyncTaskTest extends SchemaLoader
{
private static final IPartitioner partirioner = new Murmur3Partitioner();
public static final String KEYSPACE1 = "DifferencerTest";
@ -65,14 +56,8 @@ public class DifferencerTest
SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD));
}
@After
public void tearDown()
{
SinkManager.clear();
}
/**
* When there is no difference between two, Differencer should respond SYNC_COMPLETE
* When there is no difference between two, LocalSyncTask should return stats with 0 difference.
*/
@Test
public void testNoDifference() throws Throwable
@ -80,26 +65,6 @@ public class DifferencerTest
final InetAddress ep1 = InetAddress.getByName("127.0.0.1");
final InetAddress ep2 = InetAddress.getByName("127.0.0.1");
SinkManager.add(new IMessageSink()
{
@SuppressWarnings("unchecked")
public MessageOut handleMessage(MessageOut message, int id, InetAddress to)
{
if (message.verb == MessagingService.Verb.REPAIR_MESSAGE)
{
RepairMessage m = (RepairMessage) message.payload;
assertEquals(RepairMessage.Type.SYNC_COMPLETE, m.messageType);
// we should see SYNC_COMPLETE
assertEquals(new NodePair(ep1, ep2), ((SyncComplete)m).nodes);
}
return null;
}
public MessageIn handleMessage(MessageIn message, int id, InetAddress to)
{
return null;
}
});
Range<Token> range = new Range<>(partirioner.getMinimumToken(), partirioner.getRandomToken());
RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), KEYSPACE1, "Standard1", range);
@ -110,10 +75,10 @@ public class DifferencerTest
// note: we reuse the same endpoint which is bogus in theory but fine here
TreeResponse r1 = new TreeResponse(ep1, tree1);
TreeResponse r2 = new TreeResponse(ep2, tree2);
Differencer diff = new Differencer(desc, r1, r2);
diff.run();
LocalSyncTask task = new LocalSyncTask(desc, r1, r2, ActiveRepairService.UNREPAIRED_SSTABLE);
task.run();
assertTrue(diff.differences.isEmpty());
assertEquals(0, task.get().numberOfDifferences);
}
@Test
@ -144,11 +109,11 @@ public class DifferencerTest
// note: we reuse the same endpoint which is bogus in theory but fine here
TreeResponse r1 = new TreeResponse(InetAddress.getByName("127.0.0.1"), tree1);
TreeResponse r2 = new TreeResponse(InetAddress.getByName("127.0.0.2"), tree2);
Differencer diff = new Differencer(desc, r1, r2);
diff.run();
LocalSyncTask task = new LocalSyncTask(desc, r1, r2, ActiveRepairService.UNREPAIRED_SSTABLE);
task.run();
// ensure that the changed range was recorded
assertEquals("Wrong differing ranges", interesting, new HashSet<>(diff.differences));
assertEquals("Wrong differing ranges", interesting.size(), task.getCurrentStat().numberOfDifferences);
}
private MerkleTree createInitialTree(RepairJobDesc desc)

71
test/unit/org/apache/cassandra/repair/RepairSessionTest.java Normal file
View File

@ -0,0 +1,71 @@
/*
* 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.io.IOException;
import java.net.InetAddress;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import com.google.common.collect.Sets;
import org.junit.Test;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDGen;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
public class RepairSessionTest
{
@Test
public void testConviction() throws Exception
{
InetAddress remote = InetAddress.getByName("127.0.0.2");
Gossiper.instance.initializeNodeUnsafe(remote, UUID.randomUUID(), 1);
// Set up RepairSession
UUID parentSessionId = UUIDGen.getTimeUUID();
UUID sessionId = UUID.randomUUID();
IPartitioner p = new Murmur3Partitioner();
Range<Token> repairRange = new Range<>(p.getToken(ByteBufferUtil.bytes(0)), p.getToken(ByteBufferUtil.bytes(100)), p);
Set<InetAddress> endpoints = Sets.newHashSet(remote);
RepairSession session = new RepairSession(parentSessionId, sessionId, repairRange, "Keyspace1", true, endpoints, ActiveRepairService.UNREPAIRED_SSTABLE, "Standard1");
// perform convict
session.convict(remote, Double.MAX_VALUE);
// RepairSession should throw ExecutorException with the cause of IOException when getting its value
try
{
session.get();
fail();
}
catch (ExecutionException ex)
{
assertEquals(IOException.class, ex.getCause().getClass());
}
}
}

88
test/unit/org/apache/cassandra/repair/messages/RepairOptionTest.java Normal file
View File

@ -0,0 +1,88 @@
/*
* 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.messages;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import com.google.common.collect.Iterables;
import org.junit.Test;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import static org.junit.Assert.*;
public class RepairOptionTest
{
@Test
public void testParseOptions()
{
IPartitioner partitioner = new Murmur3Partitioner();
Token.TokenFactory tokenFactory = partitioner.getTokenFactory();
// parse with empty options
RepairOption option = RepairOption.parse(new HashMap<String, String>(), partitioner);
assertTrue(option.isSequential());
assertFalse(option.isPrimaryRange());
assertFalse(option.isIncremental());
// parse everything
Map<String, String> options = new HashMap<>();
options.put(RepairOption.SEQUENTIAL_KEY, "false");
options.put(RepairOption.PRIMARY_RANGE_KEY, "false");
options.put(RepairOption.INCREMENTAL_KEY, "true");
options.put(RepairOption.RANGES_KEY, "0:10,11:20,21:30");
options.put(RepairOption.COLUMNFAMILIES_KEY, "cf1,cf2,cf3");
options.put(RepairOption.DATACENTERS_KEY, "dc1,dc2,dc3");
options.put(RepairOption.HOSTS_KEY, "127.0.0.1,127.0.0.2,127.0.0.3");
option = RepairOption.parse(options, partitioner);
assertFalse(option.isSequential());
assertFalse(option.isPrimaryRange());
assertTrue(option.isIncremental());
Set<Range<Token>> expectedRanges = new HashSet<>(3);
expectedRanges.add(new Range<>(tokenFactory.fromString("0"), tokenFactory.fromString("10")));
expectedRanges.add(new Range<>(tokenFactory.fromString("11"), tokenFactory.fromString("20")));
expectedRanges.add(new Range<>(tokenFactory.fromString("21"), tokenFactory.fromString("30")));
assertEquals(expectedRanges, option.getRanges());
Set<String> expectedCFs = new HashSet<>(3);
expectedCFs.add("cf1");
expectedCFs.add("cf2");
expectedCFs.add("cf3");
assertEquals(expectedCFs, option.getColumnFamilies());
Set<String> expectedDCs = new HashSet<>(3);
expectedDCs.add("dc1");
expectedDCs.add("dc2");
expectedDCs.add("dc3");
assertEquals(expectedDCs, option.getDataCenters());
Set<String> expectedHosts = new HashSet<>(3);
expectedHosts.add("127.0.0.1");
expectedHosts.add("127.0.0.2");
expectedHosts.add("127.0.0.3");
assertEquals(expectedHosts, option.getHosts());
}
}

142
test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java → test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java
View File

@ -20,60 +20,39 @@ package org.apache.cassandra.service;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.Sets;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.IMutation;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.locator.SimpleStrategy;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.RepairJobDesc;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.utils.FBUtilities;
import static org.junit.Assert.assertEquals;
public abstract class AntiEntropyServiceTestAbstract
public class ActiveRepairServiceTest
{
// keyspace and column family to test against
public ActiveRepairService aes;
public String keyspaceName;
public String cfname;
public RepairJobDesc desc;
public ColumnFamilyStore store;
public InetAddress LOCAL, REMOTE;
public Range<Token> local_range;
private boolean initialized;
public abstract void init();
public abstract List<IMutation> getWriteData();
public static final String KEYSPACE5 = "Keyspace5";
public static final String CF_STANDRAD1 = "Standard1";
public static final String CF_COUNTER = "Counter1";
public String cfname;
public ColumnFamilyStore store;
public InetAddress LOCAL, REMOTE;
private boolean initialized;
@BeforeClass
public static void defineSchema() throws ConfigurationException
{
@ -93,57 +72,29 @@ public abstract class AntiEntropyServiceTestAbstract
SchemaLoader.startGossiper();
initialized = true;
init();
LOCAL = FBUtilities.getBroadcastAddress();
// generate a fake endpoint for which we can spoof receiving/sending trees
REMOTE = InetAddress.getByName("127.0.0.2");
store = null;
for (ColumnFamilyStore cfs : Keyspace.open(keyspaceName).getColumnFamilyStores())
{
if (cfs.name.equals(cfname))
{
store = cfs;
break;
}
}
assert store != null : "CF not found: " + cfname;
}
aes = ActiveRepairService.instance;
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
tmd.clearUnsafe();
StorageService.instance.setTokens(Collections.singleton(StorageService.getPartitioner().getRandomToken()));
tmd.updateNormalToken(StorageService.getPartitioner().getMinimumToken(), REMOTE);
assert tmd.isMember(REMOTE);
MessagingService.instance().setVersion(REMOTE, MessagingService.current_version);
Gossiper.instance.initializeNodeUnsafe(REMOTE, UUID.randomUUID(), 1);
local_range = StorageService.instance.getPrimaryRangesForEndpoint(keyspaceName, LOCAL).iterator().next();
desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), keyspaceName, cfname, local_range);
// Set a fake session corresponding to this fake request
ActiveRepairService.instance.submitArtificialRepairSession(desc);
}
@After
public void teardown() throws Exception
{
flushAES();
}
@Test
public void testGetNeighborsPlusOne() throws Throwable
{
// generate rf+1 nodes, and ensure that all nodes are returned
Set<InetAddress> expected = addTokens(1 + Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
Set<InetAddress> expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
expected.remove(FBUtilities.getBroadcastAddress());
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
Set<InetAddress> neighbors = new HashSet<InetAddress>();
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
Set<InetAddress> neighbors = new HashSet<>();
for (Range<Token> range : ranges)
{
neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, null, null));
neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, range, null, null));
}
assertEquals(expected, neighbors);
}
@ -154,19 +105,19 @@ public abstract class AntiEntropyServiceTestAbstract
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
// generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(keyspaceName).getReplicationStrategy();
Set<InetAddress> expected = new HashSet<InetAddress>();
addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy();
Set<InetAddress> expected = new HashSet<>();
for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
{
expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
}
expected.remove(FBUtilities.getBroadcastAddress());
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
Set<InetAddress> neighbors = new HashSet<InetAddress>();
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
Set<InetAddress> neighbors = new HashSet<>();
for (Range<Token> range : ranges)
{
neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, null, null));
neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, range, null, null));
}
assertEquals(expected, neighbors);
}
@ -175,20 +126,20 @@ public abstract class AntiEntropyServiceTestAbstract
public void testGetNeighborsPlusOneInLocalDC() throws Throwable
{
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
// generate rf+1 nodes, and ensure that all nodes are returned
Set<InetAddress> expected = addTokens(1 + Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
Set<InetAddress> expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
expected.remove(FBUtilities.getBroadcastAddress());
// remove remote endpoints
TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology();
HashSet<InetAddress> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
expected = Sets.intersection(expected, localEndpoints);
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
Set<InetAddress> neighbors = new HashSet<InetAddress>();
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
Set<InetAddress> neighbors = new HashSet<>();
for (Range<Token> range : ranges)
{
neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
}
assertEquals(expected, neighbors);
}
@ -199,9 +150,9 @@ public abstract class AntiEntropyServiceTestAbstract
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
// generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(keyspaceName).getReplicationStrategy();
Set<InetAddress> expected = new HashSet<InetAddress>();
addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy();
Set<InetAddress> expected = new HashSet<>();
for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
{
expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange));
@ -211,12 +162,12 @@ public abstract class AntiEntropyServiceTestAbstract
TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology();
HashSet<InetAddress> localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter()));
expected = Sets.intersection(expected, localEndpoints);
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(keyspaceName);
Set<InetAddress> neighbors = new HashSet<InetAddress>();
Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(KEYSPACE5);
Set<InetAddress> neighbors = new HashSet<>();
for (Range<Token> range : ranges)
{
neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
neighbors.addAll(ActiveRepairService.getNeighbors(KEYSPACE5, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null));
}
assertEquals(expected, neighbors);
}
@ -227,8 +178,8 @@ public abstract class AntiEntropyServiceTestAbstract
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
// generate rf*2 nodes, and ensure that only neighbors specified by the hosts are returned
addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(keyspaceName).getReplicationStrategy();
addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy();
List<InetAddress> expected = new ArrayList<>();
for (Range<Token> replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress()))
{
@ -238,22 +189,24 @@ public abstract class AntiEntropyServiceTestAbstract
expected.remove(FBUtilities.getBroadcastAddress());
Collection<String> hosts = Arrays.asList(FBUtilities.getBroadcastAddress().getCanonicalHostName(),expected.get(0).getCanonicalHostName());
assertEquals(expected.get(0), ActiveRepairService.getNeighbors(keyspaceName, StorageService.instance.getLocalRanges(keyspaceName).iterator().next(), null, hosts).iterator().next());
assertEquals(expected.get(0), ActiveRepairService.getNeighbors(KEYSPACE5,
StorageService.instance.getLocalRanges(KEYSPACE5).iterator().next(),
null, hosts).iterator().next());
}
@Test(expected = IllegalArgumentException.class)
public void testGetNeighborsSpecifiedHostsWithNoLocalHost() throws Throwable
{
addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor());
addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor());
//Dont give local endpoint
Collection<String> hosts = Arrays.asList("127.0.0.3");
ActiveRepairService.getNeighbors(keyspaceName, StorageService.instance.getLocalRanges(keyspaceName).iterator().next(), null, hosts);
ActiveRepairService.getNeighbors(KEYSPACE5, StorageService.instance.getLocalRanges(KEYSPACE5).iterator().next(), null, hosts);
}
Set<InetAddress> addTokens(int max) throws Throwable
{
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
Set<InetAddress> endpoints = new HashSet<InetAddress>();
Set<InetAddress> endpoints = new HashSet<>();
for (int i = 1; i <= max; i++)
{
InetAddress endpoint = InetAddress.getByName("127.0.0." + i);
@ -262,21 +215,4 @@ public abstract class AntiEntropyServiceTestAbstract
}
return endpoints;
}
void flushAES() throws Exception
{
final ExecutorService stage = StageManager.getStage(Stage.ANTI_ENTROPY);
final Callable noop = new Callable<Object>()
{
public Boolean call()
{
return true;
}
};
// send two tasks through the stage: one to follow existing tasks and a second to follow tasks created by
// those existing tasks: tasks won't recursively create more tasks
stage.submit(noop).get(5000, TimeUnit.MILLISECONDS);
stage.submit(noop).get(5000, TimeUnit.MILLISECONDS);
}
}

47
test/unit/org/apache/cassandra/service/AntiEntropyServiceCounterTest.java
View File

@ -1,47 +0,0 @@
package org.apache.cassandra.service;
/*
*
* 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.
*
*/
import java.util.List;
import java.util.LinkedList;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.composites.CellNames;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.db.ConsistencyLevel;
public class AntiEntropyServiceCounterTest extends AntiEntropyServiceTestAbstract
{
public void init()
{
keyspaceName = AntiEntropyServiceTestAbstract.KEYSPACE5;
cfname = AntiEntropyServiceTestAbstract.CF_COUNTER;;
}
public List<IMutation> getWriteData()
{
List<IMutation> rms = new LinkedList<IMutation>();
Mutation rm = new Mutation(keyspaceName, ByteBufferUtil.bytes("key1"));
rm.addCounter(cfname, CellNames.simpleDense(ByteBufferUtil.bytes("Column1")), 42);
rms.add(new CounterMutation(rm, ConsistencyLevel.ONE));
return rms;
}
}

47
test/unit/org/apache/cassandra/service/AntiEntropyServiceStandardTest.java
View File

@ -1,47 +0,0 @@
package org.apache.cassandra.service;
/*
*
* 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.
*
*/
import java.util.List;
import java.util.LinkedList;
import org.apache.cassandra.Util;
import org.apache.cassandra.db.*;
import org.apache.cassandra.utils.ByteBufferUtil;
public class AntiEntropyServiceStandardTest extends AntiEntropyServiceTestAbstract
{
public void init()
{
keyspaceName = AntiEntropyServiceStandardTest.KEYSPACE5;
cfname = AntiEntropyServiceStandardTest.CF_STANDRAD1;
}
public List<IMutation> getWriteData()
{
List<IMutation> rms = new LinkedList<IMutation>();
Mutation rm;
rm = new Mutation(keyspaceName, ByteBufferUtil.bytes("key1"));
rm.add(cfname, Util.cellname("Column1"), ByteBufferUtil.bytes("asdfasdf"), 0);
rms.add(rm);
return rms;
}
}