diff --git a/CHANGES.txt b/CHANGES.txt index 432299aeb9..6d67e2eac9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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 diff --git a/src/java/org/apache/cassandra/repair/Differencer.java b/src/java/org/apache/cassandra/repair/Differencer.java deleted file mode 100644 index 214d2c9a1d..0000000000 --- a/src/java/org/apache/cassandra/repair/Differencer.java +++ /dev/null @@ -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> 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 "#" + r2.endpoint + ":" + desc.columnFamily + "@" + desc.range + ">"; - } -} diff --git a/src/java/org/apache/cassandra/repair/LocalSyncTask.java b/src/java/org/apache/cassandra/repair/LocalSyncTask.java new file mode 100644 index 0000000000..38f63ce972 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/LocalSyncTask.java @@ -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> 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); + } +} diff --git a/src/java/org/apache/cassandra/repair/RemoteSyncTask.java b/src/java/org/apache/cassandra/repair/RemoteSyncTask.java new file mode 100644 index 0000000000..ca5c99862d --- /dev/null +++ b/src/java/org/apache/cassandra/repair/RemoteSyncTask.java @@ -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> 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))); + } + } +} diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java b/src/java/org/apache/cassandra/repair/RepairJob.java index 8057ed53f5..b0d17ab228 100644 --- a/src/java/org/apache/cassandra/repair/RepairJob.java +++ b/src/java/org/apache/cassandra/repair/RepairJob.java @@ -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 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 treeRequests; - // tree responses are then tracked here - private final List 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 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(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 endpoints) - { - // send requests to all nodes - List allEndpoints = new ArrayList<>(endpoints); + List allEndpoints = new ArrayList<>(session.endpoints); allEndpoints.add(FBUtilities.getBroadcastAddress()); + ListenableFuture> validations; if (isSequential) { + // Request snapshot to all replica List> 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> allSnapshotTasks = Futures.allAsList(snapshotTasks); - // Execute send tree request after all snapshot complete - Futures.addCallback(allSnapshotTasks, new FutureCallback>() + validations = Futures.transform(allSnapshotTasks, new AsyncFunction, List>() { - public void onSuccess(List endpoints) + public ListenableFuture> apply(List 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 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 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> syncResults = Futures.transform(validations, new AsyncFunction, List>() { - TreeResponse r1 = trees.get(i); - for (int j = i + 1; j < trees.size(); ++j) + public ListenableFuture> apply(List 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 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>() + { + public void onSuccess(List 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> sendValidationRequest(Collection 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> 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); } } diff --git a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java index 54117a3d92..04a27af0a8 100644 --- a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java +++ b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java @@ -103,7 +103,12 @@ public class RepairMessageVerbHandler implements IVerbHandler 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; diff --git a/src/java/org/apache/cassandra/repair/RepairFuture.java b/src/java/org/apache/cassandra/repair/RepairResult.java similarity index 77% rename from src/java/org/apache/cassandra/repair/RepairFuture.java rename to src/java/org/apache/cassandra/repair/RepairResult.java index 127d8733b3..259d5f3911 100644 --- a/src/java/org/apache/cassandra/repair/RepairFuture.java +++ b/src/java/org/apache/cassandra/repair/RepairResult.java @@ -17,15 +17,16 @@ */ package org.apache.cassandra.repair; -import java.util.concurrent.FutureTask; +import java.util.List; -public class RepairFuture extends FutureTask +public class RepairResult { - public final RepairSession session; + public final RepairJobDesc desc; + public final List stats; - public RepairFuture(RepairSession session) + public RepairResult(RepairJobDesc desc, List stats) { - super(session, null); - this.session = session; + this.desc = desc; + this.stats = stats; } } diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java index 9274342a2f..240a21cf89 100644 --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@ -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 + *
    + *
  1. 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 + *
    2. + *
  2. 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. + *
    3. + *
+ * 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> 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 range; public final Set 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 jobs = new ConcurrentLinkedQueue<>(); - - // and after receiving all validation, the job is moved to - // this map, keyed by CF name. - final Map syncingJobs = new ConcurrentHashMap<>(); + // Each validation task waits response from replica in validating ConcurrentMap (keyed by CF name and endpoint address) + private final ConcurrentMap, ValidationTask> validating = new ConcurrentHashMap<>(); + // Remote syncing jobs wait response in syncingTasks map + private final ConcurrentMap, 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 range, String keyspace, boolean isSequential, Set endpoints, String... cfnames) + public RepairSession(UUID parentRepairSession, + UUID id, + Range range, + String keyspace, + boolean isSequential, + Set 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 range, String keyspace, boolean isSequential, Set 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 key, ValidationTask task) + { + validating.put(key, task); + } + + public void waitForSync(Pair 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.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> 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>() + { + public void onSuccess(List 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); } } diff --git a/src/java/org/apache/cassandra/repair/RequestCoordinator.java b/src/java/org/apache/cassandra/repair/RequestCoordinator.java deleted file mode 100644 index ed089efce3..0000000000 --- a/src/java/org/apache/cassandra/repair/RequestCoordinator.java +++ /dev/null @@ -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 -{ - private final Order 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 - { - protected final RequestCoordinator coordinator; - - Order(RequestCoordinator coordinator) - { - this.coordinator = coordinator; - } - - public abstract void add(R request); - public abstract void start(); - public abstract int completed(R request); - } - - private static class SequentialOrder extends Order - { - private final Queue requests = new LinkedList<>(); - - SequentialOrder(RequestCoordinator 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 extends Order - { - private final Set requests = new HashSet<>(); - - ParallelOrder(RequestCoordinator 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(); - } - } - -} diff --git a/src/java/org/apache/cassandra/repair/SnapshotTask.java b/src/java/org/apache/cassandra/repair/SnapshotTask.java index 6c3afb138e..a87643c50b 100644 --- a/src/java/org/apache/cassandra/repair/SnapshotTask.java +++ b/src/java/org/apache/cassandra/repair/SnapshotTask.java @@ -74,6 +74,7 @@ public class SnapshotTask extends AbstractFuture implements Runnabl public void onFailure(InetAddress from) { + //listener.failedSnapshot(); task.setException(new RuntimeException("Could not create snapshot at " + from)); } } diff --git a/src/java/org/apache/cassandra/repair/StreamingRepairTask.java b/src/java/org/apache/cassandra/repair/StreamingRepairTask.java index 9af949da02..f30eb6fca6 100644 --- a/src/java/org/apache/cassandra/repair/StreamingRepairTask.java +++ b/src/java/org/apache/cassandra/repair/StreamingRepairTask.java @@ -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) { diff --git a/src/java/org/apache/cassandra/repair/IRepairJobEventListener.java b/src/java/org/apache/cassandra/repair/SyncStat.java similarity index 73% rename from src/java/org/apache/cassandra/repair/IRepairJobEventListener.java rename to src/java/org/apache/cassandra/repair/SyncStat.java index 778c09d6b4..5721a2083b 100644 --- a/src/java/org/apache/cassandra/repair/IRepairJobEventListener.java +++ b/src/java/org/apache/cassandra/repair/SyncStat.java @@ -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; + } } diff --git a/src/java/org/apache/cassandra/repair/SyncTask.java b/src/java/org/apache/cassandra/repair/SyncTask.java new file mode 100644 index 0000000000..3ce553211f --- /dev/null +++ b/src/java/org/apache/cassandra/repair/SyncTask.java @@ -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 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> 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> differences); +} diff --git a/src/java/org/apache/cassandra/repair/ValidationTask.java b/src/java/org/apache/cassandra/repair/ValidationTask.java new file mode 100644 index 0000000000..a52ec4f970 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/ValidationTask.java @@ -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 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)); + } + } +} diff --git a/src/java/org/apache/cassandra/repair/messages/RepairOption.java b/src/java/org/apache/cassandra/repair/messages/RepairOption.java new file mode 100644 index 0000000000..ca02365894 --- /dev/null +++ b/src/java/org/apache/cassandra/repair/messages/RepairOption.java @@ -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. + *

+ * Available options are: + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + * + *
keyvaluedefault (when key not given)
sequential"true" if perform sequential repair.true
primaryRange"true" if perform repair only on primary range.false
incremental"true" if perform incremental repair.false
jobThreadsNumber of threads to use to run repair job.1
rangesRanges to repair. A range is expressed as <start token>:<end token> + * and multiple ranges can be given as comma separated ranges(e.g. aaa:bbb,ccc:ddd).
columnFamiliesSpecify names of ColumnFamilies to repair. + * Multiple ColumnFamilies can be given as comma separated values(e.g. cf1,cf2,cf3).
dataCentersSpecify names of data centers who participate in this repair. + * Multiple data centers can be given as comma separated values(e.g. dc1,dc2,dc3).
hostsSpecify names of hosts who participate in this repair. + * Multiple hosts can be given as comma separated values(e.g. cass1,cass2).
+ * + * @param options options to parse + * @param partitioner partitioner is used to construct token ranges + * @return RepairOptions object + */ + public static RepairOption parse(Map 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> 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 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 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 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 columnFamilies = new HashSet<>(); + private final Collection dataCenters = new HashSet<>(); + private final Collection hosts = new HashSet<>(); + private final Collection> ranges = new HashSet<>(); + + public RepairOption(boolean sequential, boolean primaryRange, boolean incremental, int jobThreads, Collection> 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 getColumnFamilies() + { + return columnFamilies; + } + + public Collection> getRanges() + { + return ranges; + } + + public Collection getDataCenters() + { + return dataCenters; + } + + public Collection 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() + + ')'; + } +} diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java index f1b9ccee76..670aa0b3e6 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@ -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(), - 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 sessions; + private final ConcurrentMap sessions = new ConcurrentHashMap<>(); - private final ConcurrentMap parentRepairSessions; + private final ConcurrentMap 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 range, String keyspace, boolean isSequential, Set endpoints, String... cfnames) + public RepairSession submitRepairSession(UUID parentRepairSession, + Range range, + String keyspace, + boolean isSequential, + Set 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 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 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 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 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 neighbors, boolean doAntiCompaction) + public void finishParentSession(UUID parentSession, Set neighbors) { try { - if (doAntiCompaction) + for (InetAddress neighbor : neighbors) { - for (InetAddress neighbor : neighbors) - { - AnticompactionRequest acr = new AnticompactionRequest(parentSession); - MessageOut req = acr.createMessage(); - MessagingService.instance().sendOneWay(req, neighbor); - } - List> futures = doAntiCompaction(parentSession); - FBUtilities.waitOnFutures(futures); + AnticompactionRequest acr = new AnticompactionRequest(parentSession); + MessageOut req = acr.createMessage(); + MessagingService.instance().sendOneWay(req, neighbor); } + List> futures = doAntiCompaction(parentSession); + FBUtilities.waitOnFutures(futures); } finally { @@ -407,7 +392,7 @@ public class ActiveRepairService this.repairedAt = repairedAt; } - public Collection getAndReferenceSSTables(UUID cfId) + public synchronized Collection getAndReferenceSSTables(UUID cfId) { Set sstables = sstableMap.get(cfId); Iterator sstableIterator = sstables.iterator(); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 5b5fa20d31..eb4c3e26a6 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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 dataCenters, Collection hosts, boolean primaryRange, boolean fullRepair, String... columnFamilies) throws IOException + public int repairAsync(String keyspace, Map repairSpec) { - Collection> 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 dataCenters, Collection hosts, Collection> 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 dataCenters, + Collection hosts, + boolean primaryRange, + boolean fullRepair, + String... columnFamilies) { - Collection> 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.>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> 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 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 dataCenters, Collection hosts, boolean fullRepair, String... columnFamilies) throws IOException + public int forceRepairRangeAsync(String beginToken, + String endToken, + String keyspaceName, + boolean isSequential, + Collection dataCenters, + Collection 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> 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> 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 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 createRepairTask(int cmd, - String keyspace, - Collection> ranges, - boolean isSequential, - boolean isLocal, - boolean fullRepair, - String... columnFamilies) + public int forceRepairAsync(String keyspace, RepairOption options) { - Set 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 createRepairTask(final int cmd, - final String keyspace, - final Collection> ranges, - final boolean isSequential, - final Collection dataCenters, - final Collection hosts, - final boolean fullRepair, - final String... columnFamilies) + private FutureTask 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 allNeighbors = new HashSet<>(); + final Set allNeighbors = new HashSet<>(); Map> rangeToNeighbors = new HashMap<>(); - for (Range range : ranges) + for (Range range : options.getRanges()) { try { - Set neighbors = ActiveRepairService.getNeighbors(keyspace, range, dataCenters, hosts); + Set 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 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 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(), + new NamedThreadFactory("Repair#" + cmd), + "internal")); + + List> 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 range : ranges) + for (Range 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>() { - 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 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() { - 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); } diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index cc54639fca..203d5dc6af 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -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 options); + + @Deprecated public int forceRepairAsync(String keyspace, boolean isSequential, Collection dataCenters, Collection 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 dataCenters, Collection 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(); diff --git a/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java b/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java index 33da3d182f..f26e43980d 100644 --- a/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java +++ b/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java @@ -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; diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 70788fdf27..8793e923a7 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -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 dataCenters, Collection hosts, boolean primaryRange, boolean fullRepair, String... columnFamilies) throws IOException + public void repairAsync(final PrintStream out, final String keyspace, Map 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 dataCenters, Collection 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 dataCenters, Collection 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 dataCenters, Collection 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(); - } - } -} diff --git a/src/java/org/apache/cassandra/tools/NodeTool.java b/src/java/org/apache/cassandra/tools/NodeTool.java index 80b0b8f989..18536bfe51 100644 --- a/src/java/org/apache/cassandra/tools/NodeTool.java +++ b/src/java/org/apache/cassandra/tools/NodeTool.java @@ -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 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 dataCenters = null; - Collection 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); diff --git a/src/java/org/apache/cassandra/tools/RepairRunner.java b/src/java/org/apache/cassandra/tools/RepairRunner.java new file mode 100644 index 0000000000..1898bb47e5 --- /dev/null +++ b/src/java/org/apache/cassandra/tools/RepairRunner.java @@ -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 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 options; + + private volatile int cmd; + private volatile boolean success; + + public RepairRunner(PrintStream out, StorageServiceMBean ssProxy, String keyspace, Map 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)); + } + } +} diff --git a/test/unit/org/apache/cassandra/repair/DifferencerTest.java b/test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java similarity index 73% rename from test/unit/org/apache/cassandra/repair/DifferencerTest.java rename to test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java index e1ff26e77c..b3d333a32e 100644 --- a/test/unit/org/apache/cassandra/repair/DifferencerTest.java +++ b/test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java @@ -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 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) diff --git a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java new file mode 100644 index 0000000000..9811fcc70d --- /dev/null +++ b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java @@ -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 repairRange = new Range<>(p.getToken(ByteBufferUtil.bytes(0)), p.getToken(ByteBufferUtil.bytes(100)), p); + Set 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()); + } + } +} diff --git a/test/unit/org/apache/cassandra/repair/messages/RepairOptionTest.java b/test/unit/org/apache/cassandra/repair/messages/RepairOptionTest.java new file mode 100644 index 0000000000..59ad8a3a1b --- /dev/null +++ b/test/unit/org/apache/cassandra/repair/messages/RepairOptionTest.java @@ -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(), partitioner); + assertTrue(option.isSequential()); + assertFalse(option.isPrimaryRange()); + assertFalse(option.isIncremental()); + + // parse everything + Map 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> 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 expectedCFs = new HashSet<>(3); + expectedCFs.add("cf1"); + expectedCFs.add("cf2"); + expectedCFs.add("cf3"); + assertEquals(expectedCFs, option.getColumnFamilies()); + + Set expectedDCs = new HashSet<>(3); + expectedDCs.add("dc1"); + expectedDCs.add("dc2"); + expectedDCs.add("dc3"); + assertEquals(expectedDCs, option.getDataCenters()); + + Set 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()); + } +} diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java similarity index 62% rename from test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java rename to test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java index efa47c10d4..dab45f91cf 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java +++ b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java @@ -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 local_range; - - private boolean initialized; - - public abstract void init(); - - public abstract List 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 expected = addTokens(1 + Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor()); + Set expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor()); expected.remove(FBUtilities.getBroadcastAddress()); - Collection> ranges = StorageService.instance.getLocalRanges(keyspaceName); - Set neighbors = new HashSet(); + Collection> ranges = StorageService.instance.getLocalRanges(KEYSPACE5); + Set neighbors = new HashSet<>(); for (Range 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 expected = new HashSet(); + addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor()); + AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy(); + Set expected = new HashSet<>(); for (Range replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress())) { expected.addAll(ars.getRangeAddresses(tmd.cloneOnlyTokenMap()).get(replicaRange)); } expected.remove(FBUtilities.getBroadcastAddress()); - Collection> ranges = StorageService.instance.getLocalRanges(keyspaceName); - Set neighbors = new HashSet(); + Collection> ranges = StorageService.instance.getLocalRanges(KEYSPACE5); + Set neighbors = new HashSet<>(); for (Range 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 expected = addTokens(1 + Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor()); + Set expected = addTokens(1 + Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor()); expected.remove(FBUtilities.getBroadcastAddress()); // remove remote endpoints TokenMetadata.Topology topology = tmd.cloneOnlyTokenMap().getTopology(); HashSet localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter())); expected = Sets.intersection(expected, localEndpoints); - Collection> ranges = StorageService.instance.getLocalRanges(keyspaceName); - Set neighbors = new HashSet(); + Collection> ranges = StorageService.instance.getLocalRanges(KEYSPACE5); + Set neighbors = new HashSet<>(); for (Range 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 expected = new HashSet(); + addTokens(2 * Keyspace.open(KEYSPACE5).getReplicationStrategy().getReplicationFactor()); + AbstractReplicationStrategy ars = Keyspace.open(KEYSPACE5).getReplicationStrategy(); + Set expected = new HashSet<>(); for (Range 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 localEndpoints = Sets.newHashSet(topology.getDatacenterEndpoints().get(DatabaseDescriptor.getLocalDataCenter())); expected = Sets.intersection(expected, localEndpoints); - - Collection> ranges = StorageService.instance.getLocalRanges(keyspaceName); - Set neighbors = new HashSet(); + + Collection> ranges = StorageService.instance.getLocalRanges(KEYSPACE5); + Set neighbors = new HashSet<>(); for (Range 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 expected = new ArrayList<>(); for (Range replicaRange : ars.getAddressRanges().get(FBUtilities.getBroadcastAddress())) { @@ -238,22 +189,24 @@ public abstract class AntiEntropyServiceTestAbstract expected.remove(FBUtilities.getBroadcastAddress()); Collection 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 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 addTokens(int max) throws Throwable { TokenMetadata tmd = StorageService.instance.getTokenMetadata(); - Set endpoints = new HashSet(); + Set 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() - { - 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); - } } diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceCounterTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceCounterTest.java deleted file mode 100644 index 3f1c3aac9b..0000000000 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceCounterTest.java +++ /dev/null @@ -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 getWriteData() - { - List rms = new LinkedList(); - 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; - } -} diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceStandardTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceStandardTest.java deleted file mode 100644 index c83a7301f9..0000000000 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceStandardTest.java +++ /dev/null @@ -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 getWriteData() - { - List rms = new LinkedList(); - Mutation rm; - rm = new Mutation(keyspaceName, ByteBufferUtil.bytes("key1")); - rm.add(cfname, Util.cellname("Column1"), ByteBufferUtil.bytes("asdfasdf"), 0); - rms.add(rm); - return rms; - } -}