diff --git a/build.xml b/build.xml
index bd1f557489..6c816ebbb5 100644
--- a/build.xml
+++ b/build.xml
@@ -541,6 +541,7 @@
+
@@ -675,6 +676,7 @@
artifactId="cassandra-parent"
version="${version}"/>
+
@@ -707,6 +709,7 @@
artifactId="cassandra-parent"
version="${version}"/>
+
diff --git a/src/java/org/apache/cassandra/repair/RepairRunnable.java b/src/java/org/apache/cassandra/repair/RepairRunnable.java
index ed7e2087bb..c673a6cd05 100644
--- a/src/java/org/apache/cassandra/repair/RepairRunnable.java
+++ b/src/java/org/apache/cassandra/repair/RepairRunnable.java
@@ -19,7 +19,12 @@ package org.apache.cassandra.repair;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -28,17 +33,18 @@ import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.*;
-
-import org.apache.cassandra.locator.EndpointsForRange;
-import org.apache.cassandra.locator.Replica;
+import com.google.common.util.concurrent.AsyncFunction;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.lang3.time.DurationFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,23 +53,26 @@ import com.codahale.metrics.Timer;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.gms.FailureDetector;
-import org.apache.cassandra.repair.consistent.SyncStatSummary;
-import org.apache.cassandra.schema.SchemaConstants;
-import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.repair.consistent.CoordinatorSession;
-import org.apache.cassandra.metrics.StorageMetrics;
+import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.Replica;
+import org.apache.cassandra.metrics.StorageMetrics;
+import org.apache.cassandra.repair.consistent.CoordinatorSession;
+import org.apache.cassandra.repair.consistent.SyncStatSummary;
import org.apache.cassandra.repair.messages.RepairOption;
+import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.ActiveRepairService;
+import org.apache.cassandra.service.ActiveRepairService.ParentRepairStatus;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageService;
@@ -81,7 +90,7 @@ import org.apache.cassandra.utils.progress.ProgressEventNotifier;
import org.apache.cassandra.utils.progress.ProgressEventType;
import org.apache.cassandra.utils.progress.ProgressListener;
-public class RepairRunnable extends WrappedRunnable implements ProgressEventNotifier
+public class RepairRunnable implements Runnable, ProgressEventNotifier
{
private static final Logger logger = LoggerFactory.getLogger(RepairRunnable.class);
@@ -91,13 +100,18 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
private final String keyspace;
private final String tag;
- private final AtomicInteger progress = new AtomicInteger();
+ private final AtomicInteger progressCounter = new AtomicInteger();
private final int totalProgress;
+ private final long creationTimeMillis = System.currentTimeMillis();
+ private final UUID parentSession = UUIDGen.getTimeUUID();
+
private final List listeners = new ArrayList<>();
private static final AtomicInteger threadCounter = new AtomicInteger(1);
+ private TraceState traceState;
+
public RepairRunnable(StorageService storageService, int cmd, RepairOption options, String keyspace)
{
this.storageService = storageService;
@@ -131,130 +145,184 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
}
}
- protected void fireErrorAndComplete(int progressCount, int totalProgress, String message)
+ public void notification(String msg)
{
- StorageMetrics.repairExceptions.inc();
- String errorMessage = String.format("Repair command #%d failed with error %s", cmd, message);
- fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progressCount, totalProgress, errorMessage));
- String completionMessage = String.format("Repair command #%d finished with error", cmd);
- fireProgressEvent(new ProgressEvent(ProgressEventType.COMPLETE, progressCount, totalProgress, completionMessage));
- recordFailure(errorMessage, completionMessage);
+ logger.info(msg);
+ fireProgressEvent(new ProgressEvent(ProgressEventType.NOTIFICATION, progressCounter.get(), totalProgress, msg));
}
-
- protected void runMayThrow() throws Exception
+ private void skip(String msg)
{
- ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.IN_PROGRESS, ImmutableList.of());
- final TraceState traceState;
- final UUID parentSession = UUIDGen.getTimeUUID();
- final String tag = "repair:" + cmd;
+ notification("Repair " + parentSession + " skipped: " + msg);
+ success(msg);
+ }
- final AtomicInteger progress = new AtomicInteger();
- final int totalProgress = 4 + options.getRanges().size(); // get valid column families, calculate neighbors, validation, prepare for repair + number of ranges to repair
+ private void success(String msg)
+ {
+ fireProgressEvent(new ProgressEvent(ProgressEventType.SUCCESS, progressCounter.get(), totalProgress, msg));
+ ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.COMPLETED,
+ ImmutableList.of(msg));
+ complete(null);
+ }
- String[] columnFamilies = options.getColumnFamilies().toArray(new String[options.getColumnFamilies().size()]);
- Iterable validColumnFamilies;
+ public void notifyError(Throwable error)
+ {
+ // exception should be ignored
+ if (error instanceof SomeRepairFailedException)
+ return;
+ logger.error("Repair {} failed:", parentSession, error);
+
+ StorageMetrics.repairExceptions.inc();
+ String errorMessage = String.format("Repair command #%d failed with error %s", cmd, error.getMessage());
+ fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progressCounter.get(), totalProgress, errorMessage));
+
+ // since this can fail, update table only after updating in-memory and notification state
+ maybeStoreParentRepairFailure(error);
+ }
+
+ private void fail(String reason)
+ {
+ if (reason == null)
+ reason = "Some repair failed";
+ String completionMessage = String.format("Repair command #%d finished with error", cmd);
+
+ // Note we rely on the first message being the reason for the failure
+ // when inspecting this state from RepairRunner.queryForCompletedRepair
+ ActiveRepairService.instance.recordRepairStatus(cmd, ParentRepairStatus.FAILED,
+ ImmutableList.of(reason, completionMessage));
+
+ complete(completionMessage);
+ }
+
+ private void complete(String msg)
+ {
+ long durationMillis = System.currentTimeMillis() - creationTimeMillis;
+ if (msg == null)
+ {
+ String duration = DurationFormatUtils.formatDurationWords(durationMillis, true, true);
+ msg = String.format("Repair command #%d finished in %s", cmd, duration);
+ }
+
+ fireProgressEvent(new ProgressEvent(ProgressEventType.COMPLETE, progressCounter.get(), totalProgress, msg));
+ logger.info(msg);
+
+ ActiveRepairService.instance.removeParentRepairSession(parentSession);
+ TraceState localState = traceState;
+ if (options.isTraced() && localState != null)
+ {
+ for (ProgressListener listener : listeners)
+ localState.removeProgressListener(listener);
+ // Because DebuggableThreadPoolExecutor#afterExecute and this callback
+ // run in a nondeterministic order (within the same thread), the
+ // TraceState may have been nulled out at this point. The TraceState
+ // should be traceState, so just set it without bothering to check if it
+ // actually was nulled out.
+ Tracing.instance.set(localState);
+ Tracing.traceRepair(msg);
+ Tracing.instance.stopSession();
+ }
+
+ Keyspace.open(keyspace).metric.repairTime.update(durationMillis, TimeUnit.MILLISECONDS);
+ }
+
+ public void run()
+ {
try
{
- validColumnFamilies = storageService.getValidColumnFamilies(false, false, keyspace, columnFamilies);
- progress.incrementAndGet();
+ runMayThrow();
}
- catch (IllegalArgumentException | IOException e)
+ catch (SkipRepairException e)
{
- logger.error("Repair {} failed:", parentSession, e);
- fireErrorAndComplete(progress.get(), totalProgress, e.getMessage());
- return;
+ skip(e.getMessage());
}
+ catch (Exception | Error e)
+ {
+ notifyError(e);
+ fail(e.getMessage());
+ }
+ }
+
+ private void runMayThrow() throws Exception
+ {
+ ActiveRepairService.instance.recordRepairStatus(cmd, ParentRepairStatus.IN_PROGRESS, ImmutableList.of());
+
+ List columnFamilies = getColumnFamilies();
+ String[] cfnames = columnFamilies.stream().map(cfs -> cfs.name).toArray(String[]::new);
+
+ this.traceState = maybeCreateTraceState(columnFamilies);
+
+ notifyStarting();
+
+ NeighborsAndRanges neighborsAndRanges = getNeighborsAndRanges();
+
+ maybeStoreParentRepairStart(cfnames);
+
+ prepare(columnFamilies, neighborsAndRanges.allNeighbors, neighborsAndRanges.force);
+
+ repair(cfnames, neighborsAndRanges);
+ }
+
+ private List getColumnFamilies() throws IOException
+ {
+ String[] columnFamilies = options.getColumnFamilies().toArray(new String[options.getColumnFamilies().size()]);
+ Iterable validColumnFamilies = storageService.getValidColumnFamilies(false, false, keyspace, columnFamilies);
+ progressCounter.incrementAndGet();
if (Iterables.isEmpty(validColumnFamilies))
- {
- String message = String.format("Empty keyspace, skipping repair: %s", keyspace);
- logger.info(message);
- fireProgressEvent(new ProgressEvent(ProgressEventType.COMPLETE, 0, 0, message));
- return;
- }
+ throw new SkipRepairException(String.format("Empty keyspace, skipping repair: %s", keyspace));
+ return Lists.newArrayList(validColumnFamilies);
+ }
- final long startTime = System.currentTimeMillis();
+ private TraceState maybeCreateTraceState(Iterable columnFamilyStores)
+ {
+ if (!options.isTraced())
+ return null;
+
+ StringBuilder cfsb = new StringBuilder();
+ for (ColumnFamilyStore cfs : columnFamilyStores)
+ cfsb.append(", ").append(cfs.keyspace.getName()).append(".").append(cfs.name);
+
+ UUID sessionId = Tracing.instance.newSession(Tracing.TraceType.REPAIR);
+ TraceState traceState = Tracing.instance.begin("repair", ImmutableMap.of("keyspace", keyspace, "columnFamilies",
+ cfsb.substring(2)));
+ traceState.enableActivityNotification(tag);
+ for (ProgressListener listener : listeners)
+ traceState.addProgressListener(listener);
+ Thread queryThread = createQueryThread(cmd, sessionId);
+ queryThread.setName("RepairTracePolling");
+ queryThread.start();
+ return traceState;
+ }
+
+ private void notifyStarting()
+ {
String message = String.format("Starting repair command #%d (%s), repairing keyspace %s with %s", cmd, parentSession, keyspace,
options);
logger.info(message);
- if (options.isTraced())
- {
- StringBuilder cfsb = new StringBuilder();
- for (ColumnFamilyStore cfs : validColumnFamilies)
- cfsb.append(", ").append(cfs.keyspace.getName()).append(".").append(cfs.name);
-
- UUID sessionId = Tracing.instance.newSession(Tracing.TraceType.REPAIR);
- traceState = Tracing.instance.begin("repair", ImmutableMap.of("keyspace", keyspace, "columnFamilies",
- cfsb.substring(2)));
- message = message + " tracing with " + sessionId;
- fireProgressEvent(new ProgressEvent(ProgressEventType.START, 0, 100, message));
- Tracing.traceRepair(message);
- traceState.enableActivityNotification(tag);
- for (ProgressListener listener : listeners)
- traceState.addProgressListener(listener);
- Thread queryThread = createQueryThread(cmd, sessionId);
- queryThread.setName("RepairTracePolling");
- queryThread.start();
- }
- else
- {
- fireProgressEvent(new ProgressEvent(ProgressEventType.START, 0, 100, message));
- traceState = null;
- }
+ Tracing.traceRepair(message);
+ fireProgressEvent(new ProgressEvent(ProgressEventType.START, 0, 100, message));
+ }
+ private NeighborsAndRanges getNeighborsAndRanges()
+ {
Set allNeighbors = new HashSet<>();
List commonRanges = new ArrayList<>();
- try
+ //pre-calculate output of getLocalReplicas and pass it to getNeighbors to increase performance and prevent
+ //calculation multiple times
+ Iterable> keyspaceLocalRanges = storageService.getLocalReplicas(keyspace).ranges();
+
+ for (Range range : options.getRanges())
{
- //pre-calculate output of getLocalReplicas and pass it to getNeighbors to increase performance and prevent
- //calculation multiple times
- Iterable> keyspaceLocalRanges = storageService.getLocalReplicas(keyspace).ranges();
+ EndpointsForRange neighbors = ActiveRepairService.getNeighbors(keyspace, keyspaceLocalRanges, range,
+ options.getDataCenters(),
+ options.getHosts());
- for (Range range : options.getRanges())
- {
- EndpointsForRange neighbors = ActiveRepairService.getNeighbors(keyspace, keyspaceLocalRanges, range,
- options.getDataCenters(),
- options.getHosts());
-
- addRangeToNeighbors(commonRanges, range, neighbors);
- allNeighbors.addAll(neighbors.endpoints());
- }
-
- progress.incrementAndGet();
- }
- catch (IllegalArgumentException e)
- {
- logger.error("Repair {} failed:", parentSession, e);
- fireErrorAndComplete(progress.get(), totalProgress, e.getMessage());
- return;
+ addRangeToNeighbors(commonRanges, range, neighbors);
+ allNeighbors.addAll(neighbors.endpoints());
}
- // Validate columnfamilies
- List columnFamilyStores = new ArrayList<>();
- try
- {
- Iterables.addAll(columnFamilyStores, validColumnFamilies);
- progress.incrementAndGet();
- }
- catch (IllegalArgumentException e)
- {
- logger.error("Repair {} failed:", parentSession, e);
- fireErrorAndComplete(progress.get(), totalProgress, e.getMessage());
- return;
- }
-
- String[] cfnames = new String[columnFamilyStores.size()];
- for (int i = 0; i < columnFamilyStores.size(); i++)
- {
- cfnames[i] = columnFamilyStores.get(i).name;
- }
-
- if (!options.isPreview())
- {
- SystemDistributedKeyspace.startParentRepair(parentSession, keyspace, cfnames, options);
- }
+ progressCounter.incrementAndGet();
boolean force = options.isForcedRepair();
@@ -264,34 +332,56 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
force = !allNeighbors.equals(actualNeighbors);
allNeighbors = actualNeighbors;
}
+ return new NeighborsAndRanges(force, allNeighbors, commonRanges);
+ }
- try (Timer.Context ctx = Keyspace.open(keyspace).metric.repairPrepareTime.time())
+ private void maybeStoreParentRepairStart(String[] cfnames)
+ {
+ if (!options.isPreview())
{
- ActiveRepairService.instance.prepareForRepair(parentSession, FBUtilities.getBroadcastAddressAndPort(), allNeighbors, options, force, columnFamilyStores);
- progress.incrementAndGet();
- }
- catch (Throwable t)
- {
- logger.error("Repair {} failed:", parentSession, t);
- if (!options.isPreview())
- {
- SystemDistributedKeyspace.failParentRepair(parentSession, t);
- }
- fireErrorAndComplete(progress.get(), totalProgress, t.getMessage());
- return;
+ SystemDistributedKeyspace.startParentRepair(parentSession, keyspace, cfnames, options);
}
+ }
+ private void maybeStoreParentRepairSuccess(Collection> successfulRanges)
+ {
+ if (!options.isPreview())
+ {
+ SystemDistributedKeyspace.successfulParentRepair(parentSession, successfulRanges);
+ }
+ }
+
+ private void maybeStoreParentRepairFailure(Throwable error)
+ {
+ if (!options.isPreview())
+ {
+ SystemDistributedKeyspace.failParentRepair(parentSession, error);
+ }
+ }
+
+ private void prepare(List columnFamilies, Set allNeighbors, boolean force)
+ {
+ try (Timer.Context ignore = Keyspace.open(keyspace).metric.repairPrepareTime.time())
+ {
+ ActiveRepairService.instance.prepareForRepair(parentSession, FBUtilities.getBroadcastAddressAndPort(), allNeighbors, options, force, columnFamilies);
+ progressCounter.incrementAndGet();
+ }
+ }
+
+ private void repair(String[] cfnames, NeighborsAndRanges neighborsAndRanges)
+ {
if (options.isPreview())
{
- previewRepair(parentSession, startTime, commonRanges, cfnames);
+ previewRepair(parentSession, creationTimeMillis, neighborsAndRanges.commonRanges, cfnames);
}
else if (options.isIncremental())
{
- incrementalRepair(parentSession, startTime, force, traceState, allNeighbors, commonRanges, cfnames);
+ incrementalRepair(parentSession, creationTimeMillis, neighborsAndRanges.force, traceState,
+ neighborsAndRanges.allNeighbors, neighborsAndRanges.commonRanges, cfnames);
}
else
{
- normalRepair(parentSession, startTime, traceState, commonRanges, cfnames);
+ normalRepair(parentSession, creationTimeMillis, traceState, neighborsAndRanges.commonRanges, cfnames);
}
}
@@ -354,7 +444,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
{
List filtered = new ArrayList<>(commonRanges.size());
- for (CommonRange commonRange: commonRanges)
+ for (CommonRange commonRange : commonRanges)
{
Set endpoints = ImmutableSet.copyOf(Iterables.filter(commonRange.endpoints, liveEndpoints::contains));
Set transEndpoints = ImmutableSet.copyOf(Iterables.filter(commonRange.transEndpoints, liveEndpoints::contains));
@@ -381,9 +471,9 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
{
// the local node also needs to be included in the set of participants, since coordinator sessions aren't persisted
Set allParticipants = ImmutableSet.builder()
- .addAll(allNeighbors)
- .add(FBUtilities.getBroadcastAddressAndPort())
- .build();
+ .addAll(allNeighbors)
+ .add(FBUtilities.getBroadcastAddressAndPort())
+ .build();
List allRanges = filterCommonRanges(commonRanges, allParticipants, forceRepair);
@@ -418,8 +508,14 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
{
try
{
+ if (results == null || results.stream().anyMatch(s -> s == null))
+ {
+ // something failed
+ fail(null);
+ return;
+ }
PreviewKind previewKind = options.getPreviewKind();
- assert previewKind != PreviewKind.NONE;
+ Preconditions.checkState(previewKind != PreviewKind.NONE, "Preview is NONE");
SyncStatSummary summary = new SyncStatSummary(true);
summary.consumeSessionResults(results);
@@ -427,50 +523,31 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
if (summary.isEmpty())
{
message = previewKind == PreviewKind.REPAIRED ? "Repaired data is in sync" : "Previewed data was in sync";
- logger.info(message);
- fireProgressEvent(new ProgressEvent(ProgressEventType.NOTIFICATION, progress.get(), totalProgress, message));
}
else
{
- message = (previewKind == PreviewKind.REPAIRED ? "Repaired data is inconsistent\n" : "Preview complete\n") + summary.toString();
- logger.info(message);
- fireProgressEvent(new ProgressEvent(ProgressEventType.NOTIFICATION, progress.get(), totalProgress, message));
+ message = (previewKind == PreviewKind.REPAIRED ? "Repaired data is inconsistent\n" : "Preview complete\n") + summary.toString();
}
+ notification(message);
- String successMessage = "Repair preview completed successfully";
- fireProgressEvent(new ProgressEvent(ProgressEventType.SUCCESS, progress.get(), totalProgress,
- successMessage));
- String completionMessage = complete();
-
- ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.COMPLETED,
- ImmutableList.of(message, successMessage, completionMessage));
+ success("Repair preview completed successfully");
}
catch (Throwable t)
{
logger.error("Error completing preview repair", t);
onFailure(t);
}
+ finally
+ {
+ executor.shutdownNow();
+ }
}
public void onFailure(Throwable t)
{
- StorageMetrics.repairExceptions.inc();
- fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, t.getMessage()));
- logger.error("Error completing preview repair", t);
- String completionMessage = complete();
- recordFailure(t.getMessage(), completionMessage);
- }
-
- private String complete()
- {
- logger.debug("Preview repair {} completed", parentSession);
-
- String duration = DurationFormatUtils.formatDurationWords(System.currentTimeMillis() - startTime,
- true, true);
- String message = String.format("Repair preview #%d finished in %s", cmd, duration);
- fireProgressEvent(new ProgressEvent(ProgressEventType.COMPLETE, progress.get(), totalProgress, message));
+ notifyError(t);
+ fail("Error completing preview repair: " + t.getMessage());
executor.shutdownNow();
- return message;
}
}, MoreExecutors.directExecutor());
}
@@ -533,22 +610,16 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
session.ranges().toString());
logger.info(message);
fireProgressEvent(new ProgressEvent(ProgressEventType.PROGRESS,
- progress.incrementAndGet(),
+ progressCounter.incrementAndGet(),
totalProgress,
message));
}
public void onFailure(Throwable t)
{
- StorageMetrics.repairExceptions.inc();
-
String message = String.format("Repair session %s for range %s failed with error %s",
session.getId(), session.ranges().toString(), t.getMessage());
- logger.error(message, t);
- fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR,
- progress.incrementAndGet(),
- totalProgress,
- message));
+ notifyError(new RuntimeException(message, t));
}
}
@@ -578,86 +649,26 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
public void onSuccess(Object result)
{
- if (!options.isPreview())
- {
- SystemDistributedKeyspace.successfulParentRepair(parentSession, successfulRanges);
- }
- final String message;
+ maybeStoreParentRepairSuccess(successfulRanges);
if (hasFailure.get())
{
- StorageMetrics.repairExceptions.inc();
- message = "Some repair failed";
- fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress,
- message));
+ fail(null);
}
else
{
- message = "Repair completed successfully";
- fireProgressEvent(new ProgressEvent(ProgressEventType.SUCCESS, progress.get(), totalProgress,
- message));
- }
- String completionMessage = repairComplete();
- if (hasFailure.get())
- {
- recordFailure(message, completionMessage);
- }
- else
- {
- ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.COMPLETED,
- ImmutableList.of(message, completionMessage));
+ success("Repair completed successfully");
}
+ executor.shutdownNow();
}
public void onFailure(Throwable t)
{
- StorageMetrics.repairExceptions.inc();
- fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, t.getMessage()));
- if (!options.isPreview())
- {
- SystemDistributedKeyspace.failParentRepair(parentSession, t);
- }
- String completionMessage = repairComplete();
- recordFailure(t.getMessage(), completionMessage);
- }
-
- private String repairComplete()
- {
- ActiveRepairService.instance.removeParentRepairSession(parentSession);
- long durationMillis = System.currentTimeMillis() - startTime;
- String duration = DurationFormatUtils.formatDurationWords(durationMillis,true, true);
- String message = String.format("Repair command #%d finished in %s", cmd, duration);
- fireProgressEvent(new ProgressEvent(ProgressEventType.COMPLETE, progress.get(), totalProgress, message));
- logger.info(message);
- if (options.isTraced() && traceState != null)
- {
- for (ProgressListener listener : listeners)
- traceState.removeProgressListener(listener);
- // Because DebuggableThreadPoolExecutor#afterExecute and this callback
- // run in a nondeterministic order (within the same thread), the
- // TraceState may have been nulled out at this point. The TraceState
- // should be traceState, so just set it without bothering to check if it
- // actually was nulled out.
- Tracing.instance.set(traceState);
- Tracing.traceRepair(message);
- Tracing.instance.stopSession();
- }
+ notifyError(t);
+ fail(t.getMessage());
executor.shutdownNow();
- Keyspace.open(keyspace).metric.repairTime.update(durationMillis, TimeUnit.MILLISECONDS);
- return message;
}
}
- private void recordFailure(String failureMessage, String completionMessage)
- {
- // Note we rely on the first message being the reason for the failure
- // when inspecting this state from RepairRunner.queryForCompletedRepair
- String failure = failureMessage == null ? "unknown failure" : failureMessage;
- String completion = completionMessage == null ? "unknown completion" : completionMessage;
-
- ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.FAILED,
- ImmutableList.of(failure, completion));
- }
-
private static void addRangeToNeighbors(List neighborRangeList, Range range, EndpointsForRange neighbors)
{
Set endpoints = neighbors.endpoints();
@@ -696,7 +707,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
ByteBuffer sessionIdBytes = ByteBufferUtil.bytes(sessionId);
InetAddressAndPort source = FBUtilities.getBroadcastAddressAndPort();
- HashSet[] seen = new HashSet[] { new HashSet<>(), new HashSet<>() };
+ HashSet[] seen = new HashSet[]{ new HashSet<>(), new HashSet<>() };
int si = 0;
UUID uuid;
@@ -741,7 +752,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
if (seen[si == 0 ? 1 : 0].contains(uuid))
continue;
String message = String.format("%s: %s", r.getInetAddress("source"), r.getString("activity"));
- fireProgressEvent(new ProgressEvent(ProgressEventType.NOTIFICATION, 0, 0, message));
+ notification(message);
}
tlast = tcur;
@@ -751,4 +762,26 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
}
}, "Repair-Runnable-" + threadCounter.incrementAndGet());
}
+
+ private static final class SkipRepairException extends RuntimeException
+ {
+ SkipRepairException(String message)
+ {
+ super(message);
+ }
+ }
+
+ private static final class NeighborsAndRanges
+ {
+ private final boolean force;
+ private final Set allNeighbors;
+ private final List commonRanges;
+
+ private NeighborsAndRanges(boolean force, Set allNeighbors, List commonRanges)
+ {
+ this.force = force;
+ this.allNeighbors = allNeighbors;
+ this.commonRanges = commonRanges;
+ }
+ }
}
diff --git a/src/java/org/apache/cassandra/repair/SomeRepairFailedException.java b/src/java/org/apache/cassandra/repair/SomeRepairFailedException.java
new file mode 100644
index 0000000000..4b077b836a
--- /dev/null
+++ b/src/java/org/apache/cassandra/repair/SomeRepairFailedException.java
@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+/**
+ * This is a special exception which states "I know something failed but I don't have access to the failure". This
+ * is mostly used to make sure the error notifications are clean and the history table has a meaningful exception.
+ *
+ * The expected behavior is that when this is thrown, this error should be ignored from history table and not used
+ * for notifications
+ */
+public class SomeRepairFailedException extends RuntimeException
+{
+ public static final SomeRepairFailedException INSTANCE = new SomeRepairFailedException();
+}
diff --git a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
index 39549bd903..d0b1f70105 100644
--- a/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
+++ b/src/java/org/apache/cassandra/repair/consistent/CoordinatorSession.java
@@ -44,6 +44,7 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.repair.RepairSessionResult;
+import org.apache.cassandra.repair.SomeRepairFailedException;
import org.apache.cassandra.repair.messages.FailSession;
import org.apache.cassandra.repair.messages.FinalizeCommit;
import org.apache.cassandra.repair.messages.FinalizePropose;
@@ -322,7 +323,7 @@ public class CoordinatorSession extends ConsistentSession
logger.debug("Incremental repair {} validation/stream phase completed in {}", sessionID, formatDuration(repairStart, finalizeStart));
}
- return Futures.immediateFailedFuture(new RuntimeException());
+ return Futures.immediateFailedFuture(SomeRepairFailedException.INSTANCE);
}
else
{
diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java
index 7499c362f0..467d2bcc38 100644
--- a/src/java/org/apache/cassandra/service/ActiveRepairService.java
+++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java
@@ -520,12 +520,15 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
public synchronized ParentRepairSession removeParentRepairSession(UUID parentSessionId)
{
String snapshotName = parentSessionId.toString();
- for (ColumnFamilyStore cfs : getParentRepairSession(parentSessionId).columnFamilyStores.values())
+ ParentRepairSession session = parentRepairSessions.remove(parentSessionId);
+ if (session == null)
+ return null;
+ for (ColumnFamilyStore cfs : session.columnFamilyStores.values())
{
if (cfs.snapshotExists(snapshotName))
cfs.clearSnapshot(snapshotName);
}
- return parentRepairSessions.remove(parentSessionId);
+ return session;
}
public void handleMessage(Message message)
diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java
index 77efb0e338..180b231e1c 100644
--- a/src/java/org/apache/cassandra/tools/NodeProbe.java
+++ b/src/java/org/apache/cassandra/tools/NodeProbe.java
@@ -180,6 +180,13 @@ public class NodeProbe implements AutoCloseable
connect();
}
+ protected NodeProbe()
+ {
+ // this constructor is only used for extensions to rewrite their own connect method
+ this.host = "";
+ this.port = 0;
+ }
+
/**
* Create a connection to the JMX agent and setup the M[X]Bean proxies.
*
@@ -920,6 +927,10 @@ public class NodeProbe implements AutoCloseable
return spProxy;
}
+ public StorageServiceMBean getStorageService() {
+ return ssProxy;
+ }
+
public GossiperMBean getGossProxy()
{
return gossProxy;
diff --git a/src/java/org/apache/cassandra/tools/NodeTool.java b/src/java/org/apache/cassandra/tools/NodeTool.java
index e29f228e95..5af3fb1268 100644
--- a/src/java/org/apache/cassandra/tools/NodeTool.java
+++ b/src/java/org/apache/cassandra/tools/NodeTool.java
@@ -276,13 +276,13 @@ public class NodeTool
}
}
- private static void badUse(Exception e)
+ protected void badUse(Exception e)
{
System.out.println("nodetool: " + e.getMessage());
System.out.println("See 'nodetool help' or 'nodetool help '.");
}
- private static void err(Throwable e)
+ protected void err(Throwable e)
{
System.err.println("error: " + e.getMessage());
System.err.println("-- StackTrace --");
diff --git a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
index ef4485318b..fe969b181c 100644
--- a/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
+++ b/test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
@@ -27,7 +27,11 @@ public interface ICoordinator
{
// a bit hacky, but ConsistencyLevel draws in too many dependent classes, so we cannot have a cross-version
// method signature that accepts ConsistencyLevel directly. So we just accept an Enum and cast.
- Object[][] execute(String query, Enum consistencyLevel, Object... boundValues);
+ default Object[][] execute(String query, Enum consistencyLevel, Object... boundValues)
+ {
+ return executeWithResult(query, consistencyLevel, boundValues).toObjectArrays();
+ }
+ QueryResult executeWithResult(String query, Enum consistencyLevel, Object... boundValues);
Iterator executeWithPaging(String query, Enum consistencyLevel, int pageSize, Object... boundValues);
diff --git a/test/distributed/org/apache/cassandra/distributed/api/IInstance.java b/test/distributed/org/apache/cassandra/distributed/api/IInstance.java
index 8e02e23af2..23ccd7c943 100644
--- a/test/distributed/org/apache/cassandra/distributed/api/IInstance.java
+++ b/test/distributed/org/apache/cassandra/distributed/api/IInstance.java
@@ -43,7 +43,14 @@ public interface IInstance extends IIsolatedExecutor
int liveMemberCount();
- int nodetool(String... commandAndArgs);
+ NodeToolResult nodetoolResult(boolean withNotifications, String... commandAndArgs);
+ default NodeToolResult nodetoolResult(String... commandAndArgs)
+ {
+ return nodetoolResult(true, commandAndArgs);
+ }
+ default int nodetool(String... commandAndArgs) {
+ return nodetoolResult(commandAndArgs).getRc();
+ }
void uncaughtException(Thread t, Throwable e);
/**
diff --git a/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java b/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java
index 01fe97282c..f2cd6ee232 100644
--- a/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java
+++ b/test/distributed/org/apache/cassandra/distributed/api/IMessageFilters.java
@@ -18,6 +18,8 @@
package org.apache.cassandra.distributed.api;
+import java.util.function.Predicate;
+
public interface IMessageFilters
{
public interface Filter
@@ -31,6 +33,21 @@ public interface IMessageFilters
Builder from(int ... nums);
Builder to(int ... nums);
+ Builder verbs(int... verbs);
+ Builder allVerbs();
+
+ Builder inbound(boolean inbound);
+
+ default Builder inbound()
+ {
+ return inbound(true);
+ }
+
+ default Builder outbound()
+ {
+ return inbound(false);
+ }
+
/**
* Every message for which matcher returns `true` will be _dropped_ (assuming all
* other matchers in the chain will return `true` as well).
@@ -42,15 +59,42 @@ public interface IMessageFilters
public interface Matcher
{
boolean matches(int from, int to, IMessage message);
+
+ static Matcher of(Predicate fn) {
+ return (from, to, m) -> fn.test(m);
+ }
}
- Builder verbs(int... verbs);
- Builder allVerbs();
+ Builder inbound(boolean inbound);
+ default Builder inbound() {
+ return inbound(true);
+ }
+ default Builder outbound() {
+ return inbound(false);
+ }
+ default Builder verbs(int... verbs) {
+ return inbound().verbs(verbs);
+ }
+ default Builder allVerbs() {
+ return inbound().allVerbs();
+ }
void reset();
/**
- * {@code true} value returned by the implementation implies that the message was
+ * Checks if the message should be delivered. This is expected to run on "inbound", or on the reciever of
+ * the message (instance.config.num == to).
+ *
+ * @return {@code true} value returned by the implementation implies that the message was
* not matched by any filters and therefore should be delivered.
*/
- boolean permit(int from, int to, IMessage msg);
+ boolean permitInbound(int from, int to, IMessage msg);
+
+ /**
+ * Checks if the message should be delivered. This is expected to run on "outbound", or on the sender of
+ * the message (instance.config.num == from).
+ *
+ * @return {@code true} value returned by the implementation implies that the message was
+ * not matched by any filters and therefore should be delivered.
+ */
+ boolean permitOutbound(int from, int to, IMessage msg);
}
diff --git a/test/distributed/org/apache/cassandra/distributed/api/LongTokenRange.java b/test/distributed/org/apache/cassandra/distributed/api/LongTokenRange.java
new file mode 100644
index 0000000000..06327e8c8c
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/api/LongTokenRange.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import java.io.Serializable;
+
+public final class LongTokenRange implements Serializable
+{
+ public final long minExclusive;
+ public final long maxInclusive;
+
+ public LongTokenRange(long minExclusive, long maxInclusive)
+ {
+ this.minExclusive = minExclusive;
+ this.maxInclusive = maxInclusive;
+ }
+
+ public String toString()
+ {
+ return "(" + minExclusive + "," + maxInclusive + "]";
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/api/NodeToolResult.java b/test/distributed/org/apache/cassandra/distributed/api/NodeToolResult.java
new file mode 100644
index 0000000000..9ba112794e
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/api/NodeToolResult.java
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import javax.management.Notification;
+
+import org.junit.Assert;
+
+public class NodeToolResult
+{
+ private final String[] commandAndArgs;
+ private final int rc;
+ private final List notifications;
+ private final Throwable error;
+
+ public NodeToolResult(String[] commandAndArgs, int rc, List notifications, Throwable error)
+ {
+ this.commandAndArgs = commandAndArgs;
+ this.rc = rc;
+ this.notifications = notifications;
+ this.error = error;
+ }
+
+ public String[] getCommandAndArgs()
+ {
+ return commandAndArgs;
+ }
+
+ public int getRc()
+ {
+ return rc;
+ }
+
+ public List getNotifications()
+ {
+ return notifications;
+ }
+
+ public Throwable getError()
+ {
+ return error;
+ }
+
+ public Asserts asserts()
+ {
+ return new Asserts();
+ }
+
+ public final class Asserts {
+ public Asserts success() {
+ Assert.assertEquals("nodetool command " + commandAndArgs[0] + " was not successful", 0, rc);
+ return this;
+ }
+
+ public Asserts failure() {
+ Assert.assertNotEquals("nodetool command " + commandAndArgs[0] + " was successful but not expected to be", 0, rc);
+ return this;
+ }
+
+ public Asserts errorContains(String msg) {
+ Assert.assertNotNull("No exception was found but expected one", error);
+ Assert.assertTrue("Error message '" + error.getMessage() + "' does not contain '" + msg + "'", error.getMessage().contains(msg));
+ return this;
+ }
+
+ public Asserts notificationContains(String msg) {
+ Assert.assertNotNull("notifications not defined", notifications);
+ Assert.assertFalse("notifications not defined", notifications.isEmpty());
+ for (Notification n : notifications) {
+ if (n.getMessage().contains(msg)) {
+ return this;
+ }
+ }
+ Assert.fail("Unable to locate message " + msg + " in notifications: " + notifications);
+ return this; // unreachable
+ }
+
+ public Asserts notificationContains(ProgressEventType type, String msg) {
+ int userType = type.ordinal();
+ Assert.assertNotNull("notifications not defined", notifications);
+ Assert.assertFalse("notifications not defined", notifications.isEmpty());
+ for (Notification n : notifications) {
+ if (notificationType(n) == userType) {
+ if (n.getMessage().contains(msg)) {
+ return this;
+ }
+ }
+ }
+ Assert.fail("Unable to locate message '" + msg + "' in notifications: " + notifications);
+ return this; // unreachable
+ }
+ }
+
+ private static int notificationType(Notification n)
+ {
+ return ((Map) n.getUserData()).get("type").intValue();
+ }
+
+ public String toString()
+ {
+ return "NodeToolResult{" +
+ "commandAndArgs=" + Arrays.toString(commandAndArgs) +
+ ", rc=" + rc +
+ ", notifications=[" + notifications.stream().map(n -> ProgressEventType.values()[notificationType(n)].name()).collect(Collectors.joining(", ")) + "]" +
+ ", error=" + error +
+ '}';
+ }
+
+ /**
+ * Progress event type.
+ *
+ *
+ * Progress starts by emitting {@link #START}, followed by emitting zero or more {@link #PROGRESS} events,
+ * then it emits either one of {@link #ERROR}/{@link #ABORT}/{@link #SUCCESS}.
+ * Progress indicates its completion by emitting {@link #COMPLETE} at the end of process.
+ *
+ *
+ * {@link #NOTIFICATION} event type is used to just notify message without progress.
+ *
+ */
+ public enum ProgressEventType
+ {
+ /**
+ * Fired first when progress starts.
+ * Happens only once.
+ */
+ START,
+
+ /**
+ * Fire when progress happens.
+ * This can be zero or more time after START.
+ */
+ PROGRESS,
+
+ /**
+ * When observing process completes with error, this is sent once before COMPLETE.
+ */
+ ERROR,
+
+ /**
+ * When observing process is aborted by user, this is sent once before COMPLETE.
+ */
+ ABORT,
+
+ /**
+ * When observing process completes successfully, this is sent once before COMPLETE.
+ */
+ SUCCESS,
+
+ /**
+ * Fire when progress complete.
+ * This is fired once, after ERROR/ABORT/SUCCESS is fired.
+ * After this, no more ProgressEvent should be fired for the same event.
+ */
+ COMPLETE,
+
+ /**
+ * Used when sending message without progress.
+ */
+ NOTIFICATION
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/api/QueryResult.java b/test/distributed/org/apache/cassandra/distributed/api/QueryResult.java
new file mode 100644
index 0000000000..dcdfa14a00
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/api/QueryResult.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.function.Predicate;
+
+/**
+ * A table of data representing a complete query result.
+ *
+ * A QueryResult is different from {@link java.sql.ResultSet} in several key ways:
+ *
+ *
+ *
represents a complete result rather than a cursor
+ *
returns a {@link Row} to access the current row of data
+ *
relies on object pooling; {@link #hasNext()} may return the same object just with different data, accessing a
+ * {@link Row} from a previous {@link #hasNext()} call has undefined behavior.
+ *
includes {@link #filter(Predicate)}, this will do client side filtering since Apache Cassandra is more
+ * restrictive on server side filtering
+ *
+ *
+ *
Unsafe patterns
+ *
+ * Below are a few unsafe patterns which may lead to unexpected results
+ *
+ * {@code
+ * while (rs.hasNext()) {
+ * list.add(rs.next());
+ * }
+ * }
+ *
+ * {@code
+ * rs.forEach(list::add)
+ * }
+ *
+ * Both cases have the same issue; reference to a row from a previous call to {@link #hasNext()}. Since the same {@link Row}
+ * object can be used accross different calls to {@link #hasNext()} this would mean any attempt to access after the fact
+ * points to newer data. If this behavior is not desirable and access is needed between calls, then {@link Row#copy()}
+ * should be used; this will clone the {@link Row} and return a new object pointing to the same data.
+ */
+public class QueryResult implements Iterator
+{
+ public static final QueryResult EMPTY = new QueryResult(new String[0], null);
+
+ private final String[] names;
+ private final Object[][] results;
+ private final Predicate filter;
+ private final Row row;
+ private int offset = -1;
+
+ public QueryResult(String[] names, Object[][] results)
+ {
+ this.names = Objects.requireNonNull(names, "names");
+ this.results = results;
+ this.row = new Row(names);
+ this.filter = ignore -> true;
+ }
+
+ private QueryResult(String[] names, Object[][] results, Predicate filter, int offset)
+ {
+ this.names = names;
+ this.results = results;
+ this.filter = filter;
+ this.offset = offset;
+ this.row = new Row(names);
+ }
+
+ public String[] getNames()
+ {
+ return names;
+ }
+
+ public boolean isEmpty()
+ {
+ return results.length == 0;
+ }
+
+ public int size()
+ {
+ return results.length;
+ }
+
+ public QueryResult filter(Predicate fn)
+ {
+ return new QueryResult(names, results, filter.and(fn), offset);
+ }
+
+ /**
+ * Get all rows as a 2d array. Any calls to {@link #filter(Predicate)} will be ignored and the array returned will
+ * be the full set from the query.
+ */
+ public Object[][] toObjectArrays()
+ {
+ return results;
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ if (results == null)
+ return false;
+ while ((offset += 1) < results.length)
+ {
+ row.setResults(results[offset]);
+ if (filter.test(row))
+ {
+ return true;
+ }
+ }
+ row.setResults(null);
+ return false;
+ }
+
+ @Override
+ public Row next()
+ {
+ if (offset < 0 || offset >= results.length)
+ throw new NoSuchElementException();
+ return row;
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/api/Row.java b/test/distributed/org/apache/cassandra/distributed/api/Row.java
new file mode 100644
index 0000000000..43fa6d998f
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/api/Row.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.api;
+
+import java.util.Arrays;
+import java.util.Date;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import javax.annotation.Nullable;
+
+import com.carrotsearch.hppc.ObjectIntHashMap;
+import com.carrotsearch.hppc.ObjectIntMap;
+
+/**
+ * Data representing a single row in a query result.
+ *
+ * This class is mutable from the parent {@link QueryResult} and can have the row it points to changed between calls
+ * to {@link QueryResult#hasNext()}, for this reason it is unsafe to hold reference to this class after that call;
+ * to get around this, a call to {@link #copy()} will return a new object pointing to the same row.
+ */
+public class Row
+{
+ private final ObjectIntMap nameIndex;
+ @Nullable private Object[] results; // mutable to avoid allocations in loops
+
+ public Row(String[] names)
+ {
+ Objects.requireNonNull(names, "names");
+ this.nameIndex = new ObjectIntHashMap<>(names.length);
+ for (int i = 0; i < names.length; i++) {
+ nameIndex.put(names[i], i);
+ }
+ }
+
+ private Row(ObjectIntMap nameIndex)
+ {
+ this.nameIndex = nameIndex;
+ }
+
+ void setResults(@Nullable Object[] results)
+ {
+ this.results = results;
+ }
+
+ /**
+ * Creates a copy of the current row; can be used past calls to {@link QueryResult#hasNext()}.
+ */
+ public Row copy() {
+ Row copy = new Row(nameIndex);
+ copy.setResults(results);
+ return copy;
+ }
+
+ public T get(String name)
+ {
+ checkAccess();
+ int idx = findIndex(name);
+ if (idx == -1)
+ return null;
+ return (T) results[idx];
+ }
+
+ public String getString(String name)
+ {
+ return get(name);
+ }
+
+ public UUID getUUID(String name)
+ {
+ return get(name);
+ }
+
+ public Date getTimestamp(String name)
+ {
+ return get(name);
+ }
+
+ public Set getSet(String name)
+ {
+ return get(name);
+ }
+
+ public String toString()
+ {
+ return "Row{" +
+ "names=" + nameIndex.keys() +
+ ", results=" + Arrays.toString(results) +
+ '}';
+ }
+
+ private void checkAccess()
+ {
+ if (results == null)
+ throw new NoSuchElementException();
+ }
+
+ private int findIndex(String name)
+ {
+ return nameIndex.getOrDefault(name, -1);
+ }
+}
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index f03bae0d4f..371de54f0c 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@ -56,6 +56,7 @@ import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.distributed.api.IListen;
import org.apache.cassandra.distributed.api.IMessage;
import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.api.NodeToolResult;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Verb;
@@ -151,9 +152,16 @@ public abstract class AbstractCluster implements ICluster,
@Override
public synchronized void startup()
{
+ startup(AbstractCluster.this);
+ }
+
+ public synchronized void startup(ICluster cluster)
+ {
+ if (cluster != AbstractCluster.this)
+ throw new IllegalArgumentException("Only the owning cluster can be used for startup"); //TODO why have this in the API?
if (!isShutdown)
throw new IllegalStateException();
- delegate().startup(AbstractCluster.this);
+ delegate().startup(cluster);
isShutdown = false;
updateMessagingVersions();
}
@@ -183,9 +191,9 @@ public abstract class AbstractCluster implements ICluster,
throw new IllegalStateException("Cannot get live member count on shutdown instance");
}
- public int nodetool(String... commandAndArgs)
+ public NodeToolResult nodetoolResult(boolean withNotifications, String... commandAndArgs)
{
- return delegate().nodetool(commandAndArgs);
+ return delegate().nodetoolResult(withNotifications, commandAndArgs);
}
public long killAttempts()
@@ -355,7 +363,7 @@ public abstract class AbstractCluster implements ICluster,
return filters;
}
- public MessageFilters.Builder verbs(Verb... verbs)
+ public IMessageFilters.Builder verbs(Verb... verbs)
{
int[] ids = new int[verbs.length];
for (int i = 0; i < verbs.length; ++i)
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
index 94c7e9ece0..dee9049db9 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
@@ -34,6 +34,7 @@ import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.QueryResult;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.pager.QueryPager;
@@ -52,9 +53,9 @@ public class Coordinator implements ICoordinator
}
@Override
- public Object[][] execute(String query, Enum consistencyLevelOrigin, Object... boundValues)
+ public QueryResult executeWithResult(String query, Enum consistencyLevel, Object... boundValues)
{
- return instance.sync(() -> executeInternal(query, consistencyLevelOrigin, boundValues)).call();
+ return instance.sync(() -> executeInternal(query, consistencyLevel, boundValues)).call();
}
public Future asyncExecuteWithTracing(UUID sessionId, String query, Enum consistencyLevelOrigin, Object... boundValues)
@@ -63,7 +64,7 @@ public class Coordinator implements ICoordinator
try
{
Tracing.instance.newSession(sessionId, Collections.emptyMap());
- return executeInternal(query, consistencyLevelOrigin, boundValues);
+ return executeInternal(query, consistencyLevelOrigin, boundValues).toObjectArrays();
}
finally
{
@@ -72,7 +73,7 @@ public class Coordinator implements ICoordinator
}).call();
}
- private Object[][] executeInternal(String query, Enum consistencyLevelOrigin, Object[] boundValues)
+ private QueryResult executeInternal(String query, Enum consistencyLevelOrigin, Object[] boundValues)
{
ClientState clientState = makeFakeClientState();
CQLStatement prepared = QueryProcessor.getStatement(query, clientState);
@@ -94,9 +95,16 @@ public class Coordinator implements ICoordinator
System.nanoTime());
if (res != null && res.kind == ResultMessage.Kind.ROWS)
- return RowUtil.toObjects((ResultMessage.Rows) res);
+ {
+ ResultMessage.Rows rows = (ResultMessage.Rows) res;
+ String[] names = rows.result.metadata.names.stream().map(c -> c.name.toString()).toArray(String[]::new);
+ Object[][] results = RowUtil.toObjects(rows);
+ return new QueryResult(names, results);
+ }
else
- return new Object[][]{};
+ {
+ return QueryResult.EMPTY;
+ }
}
public Object[][] executeWithTracing(UUID sessionId, String query, Enum consistencyLevelOrigin, Object... boundValues)
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
index 1beb708b8e..90d747ef9d 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java
@@ -25,10 +25,13 @@ import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
-import java.util.function.BiConsumer;
+import javax.management.ListenerNotFoundException;
+import javax.management.Notification;
+import javax.management.NotificationListener;
import com.google.common.annotations.VisibleForTesting;
@@ -58,6 +61,8 @@ import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.IListen;
import org.apache.cassandra.distributed.api.IMessage;
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbe;
import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbeFactory;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.Gossiper;
@@ -81,6 +86,7 @@ import org.apache.cassandra.service.DefaultFSErrorHandler;
import org.apache.cassandra.service.PendingRangeCalculatorService;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.service.StorageServiceMBean;
import org.apache.cassandra.streaming.StreamReceiveTask;
import org.apache.cassandra.streaming.StreamTransferTask;
import org.apache.cassandra.streaming.async.StreamingInboundHandler;
@@ -199,31 +205,37 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
private void registerMockMessaging(ICluster cluster)
{
- BiConsumer deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message);
- BiConsumer deliverToInstanceIfNotFiltered = (to, message) -> {
- if (permitMessage(cluster, to, message))
- deliverToInstance.accept(to, message);
- };
-
MessagingService.instance().outboundSink.add((message, to) -> {
- deliverToInstanceIfNotFiltered.accept(to, serializeMessage(message.from(), to, message));
+ cluster.get(to).receiveMessage(serializeMessage(message.from(), to, message));
return false;
});
}
- // unnecessary if registerMockMessaging used
- private void registerFilter(ICluster cluster)
+ private void registerInboundFilter(ICluster cluster)
{
- MessagingService.instance().outboundSink.add((message, to) -> {
- return permitMessage(cluster, to, serializeMessage(message.from(), to, message));
- });
+ MessagingService.instance().inboundSink.add(message ->
+ permitMessageInbound(cluster, serializeMessage(message.from(), broadcastAddressAndPort(), message)));
}
- private boolean permitMessage(ICluster cluster, InetAddressAndPort to, IMessage message)
+ private void registerOutboundFilter(ICluster cluster)
+ {
+
+ MessagingService.instance().outboundSink.add((message, to) ->
+ permitMessageOutbound(cluster, to, serializeMessage(message.from(), to, message)));
+ }
+
+ private boolean permitMessageInbound(ICluster cluster, IMessage message)
{
int fromNum = cluster.get(message.from()).config().num();
+ int toNum = config.num(); // since this instance is reciving the message, to will always be this instance
+ return cluster.filters().permitInbound(fromNum, toNum, message);
+ }
+
+ private boolean permitMessageOutbound(ICluster cluster, InetAddressAndPort to, IMessage message)
+ {
+ int fromNum = config.num(); // since this instance is sending the message, from will always be this instance
int toNum = cluster.get(to).config().num();
- return cluster.filters().permit(fromNum, toNum, message);
+ return cluster.filters().permitOutbound(fromNum, toNum, message);
}
public void uncaughtException(Thread thread, Throwable throwable)
@@ -274,8 +286,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
Message.Header header = messageIn.header;
TraceState state = Tracing.instance.initializeFromMessage(header);
if (state != null) state.trace("{} message received from {}", header.verb, header.from);
- header.verb.stage.execute(ThrowingRunnable.toRunnable(() -> messageIn.verb().handler().doVerb((Message