diff --git a/CHANGES.txt b/CHANGES.txt index c82c8ee7dc..fc06bde5a5 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0 + * Add repair streaming preview (CASSANDRA-13257) * Cleanup isIncremental/repairedAt usage (CASSANDRA-13430) * Change protocol to allow sending key space independent of query string (CASSANDRA-10145) * Make gc_log and gc_warn settable at runtime (CASSANDRA-12661) diff --git a/NEWS.txt b/NEWS.txt index 1ec463773d..1c456b5558 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -23,6 +23,8 @@ New features - Support for arithmetic operations between `timestamp`/`date` and `duration` has been added. See CASSANDRA-11936 - Support for arithmetic operations on number has been added. See CASSANDRA-11935 + - Preview expected streaming required for a repair (nodetool repair --preview), and validate the + consistency of repaired data between nodes (nodetool repair --validate). See CASSANDRA-13257 Upgrading --------- @@ -46,7 +48,9 @@ Upgrading data to be inconsistent between nodes. The fix changes the behavior of both full and incremental repairs. For full repairs, data is no longer marked repaired. For incremental repairs, anticompaction is run at the beginning - of the repair, instead of at the end. + of the repair, instead of at the end. If incremental repair was being used + prior to upgrading, a full repair should be run after upgrading to resolve + any inconsistencies. - Config option index_interval has been removed (it was deprecated since 2.0) - Deprecated repair JMX APIs are removed. diff --git a/doc/gen-nodetool-docs.py b/doc/gen-nodetool-docs.py index ae0153476c..e3862f73b3 100644 --- a/doc/gen-nodetool-docs.py +++ b/doc/gen-nodetool-docs.py @@ -27,7 +27,7 @@ nodetool = "../bin/nodetool" outdir = "source/tools/nodetool" helpfilename = outdir + "/nodetool.txt" command_re = re.compile("( )([_a-z]+)") -commandRSTContent = ".. _{0}\n\n{0}\n-------\n\nUsage\n---------\n\n.. include:: {0}.txt\n :literal:\n\n" +commandRSTContent = ".. _nodetool_{0}:\n\n{0}\n-------\n\nUsage\n---------\n\n.. include:: {0}.txt\n :literal:\n\n" # create the documentation directory if not os.path.exists(outdir): diff --git a/doc/source/operating/repair.rst b/doc/source/operating/repair.rst index 97d8ce8bad..97115dc66b 100644 --- a/doc/source/operating/repair.rst +++ b/doc/source/operating/repair.rst @@ -16,7 +16,92 @@ .. highlight:: none +.. _repair: + Repair ------ -.. todo:: todo +Cassandra is designed to remain available if one of it's nodes is down or unreachable. However, when a node is down or +unreachable, it needs to eventually discover the writes it missed. Hints attempt to inform a node of missed writes, but +are a best effort, and aren't guaranteed to inform a node of 100% of the writes it missed. These inconsistencies can +eventually result in data loss as nodes are replaced or tombstones expire. + +These inconsistencies are fixed with the repair process. Repair synchronizes the data between nodes by comparing their +respective datasets for their common token ranges, and streaming the differences for any out of sync sections between +the nodes. It compares the data with merkle trees, which are a hierarchy of hashes. + +Incremental and Full Repairs +^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +There are 2 types of repairs: full repairs, and incremental repairs. Full repairs operate over all of the data in the +token range being repaired. Incremental repairs only repair data that's been written since the previous incremental repair. + +Incremental repairs are the default repair type, and if run regularly, can significantly reduce the time and io cost of +performing a repair. However, it's important to understand that once an incremental repair marks data as repaired, it won't +try to repair it again. This is fine for syncing up missed writes, but it doesn't protect against things like disk corruption, +data loss by operator error, or bugs in Cassandra. For this reason, full repairs should still be run occasionally. + +Usage and Best Practices +^^^^^^^^^^^^^^^^^^^^^^^^ + +Since repair can result in a lot of disk and network io, it's not run automatically by Cassandra. It is run by the operator +via nodetool. + +Incremental repair is the default and is run with the following command: + +:: + + nodetool repair + +A full repair can be run with the following command: + +:: + + nodetool repair --full + +Additionally, repair can be run on a single keyspace: + +:: + + nodetool repair [options] + +Or even on specific tables: + +:: + + nodetool repair [options] + + +The repair command only repairs token ranges on the node being repaired, it doesn't repair the whole cluster. By default, repair +will operate on all token ranges replicated by the node you're running repair on, which will cause duplicate work if you run it +on every node. The ``-pr`` flag will only repair the "primary" ranges on a node, so you can repair your entire cluster by running +``nodetool repair -pr`` on each node in a single datacenter. + +The specific frequency of repair that's right for your cluster, of course, depends on several factors. However, if you're +just starting out and looking for somewhere to start, running an incremental repair every 1-3 days, and a full repair every +1-3 weeks is probably reasonable. If you don't want to run incremental repairs, a full repair every 5 days is a good place +to start. + +At a minimum, repair should be run often enough that the gc grace period never expires on unrepaired data. Otherwise, deleted +data could reappear. With a default gc grace period of 10 days, repairing every node in your cluster at least once every 7 days +will prevent this, while providing enough slack to allow for delays. + +Other Options +^^^^^^^^^^^^^ + +``-pr, --partitioner-range`` + Restricts repair to the 'primary' token ranges of the node being repaired. A primary range is just a token range for + which a node is the first replica in the ring. + +``-prv, --preview`` + Estimates the amount of streaming that would occur for the given repair command. This builds the merkle trees, and prints + the expected streaming activity, but does not actually do any streaming. By default, incremental repairs are estimated, + add the ``--full`` flag to estimate a full repair. + +``-vd, --validate`` + Verifies that the repaired data is the same across all nodes. Similiar to ``--preview``, this builds and compares merkle + trees of repaired data, but doesn't do any streaming. This is useful for troubleshooting. If this shows that the repaired + data is out of sync, a full repair should be run. + +.. seealso:: + :ref:`nodetool repair docs ` diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index a3ca86aa23..6f311a263d 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -29,6 +29,7 @@ import javax.management.openmbean.OpenDataException; import javax.management.openmbean.TabularData; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; import com.google.common.collect.*; import com.google.common.util.concurrent.*; import org.slf4j.Logger; @@ -69,6 +70,7 @@ import org.apache.cassandra.repair.Validator; import org.apache.cassandra.schema.CompactionParams.TombstoneOption; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.utils.*; import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.concurrent.Refs; @@ -611,8 +613,11 @@ public class CompactionManager implements CompactionManagerMBean UUID pendingRepair, UUID parentRepairSession) throws InterruptedException, IOException { - logger.info("[repair #{}] Starting anticompaction for {}.{} on {}/{} sstables", parentRepairSession, cfs.keyspace.getName(), cfs.getTableName(), validatedForRepair.size(), cfs.getLiveSSTables()); - logger.trace("[repair #{}] Starting anticompaction for ranges {}", parentRepairSession, ranges); + ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(parentRepairSession); + Preconditions.checkArgument(!prs.isPreview(), "Cannot anticompact for previews"); + + logger.info("{} Starting anticompaction for {}.{} on {}/{} sstables", PreviewKind.NONE.logPrefix(parentRepairSession), cfs.keyspace.getName(), cfs.getTableName(), validatedForRepair.size(), cfs.getLiveSSTables()); + logger.trace("{} Starting anticompaction for ranges {}", PreviewKind.NONE.logPrefix(parentRepairSession), ranges); Set sstables = new HashSet<>(validatedForRepair); Set mutatedRepairStatuses = new HashSet<>(); // we should only notify that repair status changed if it actually did: @@ -640,7 +645,7 @@ public class CompactionManager implements CompactionManagerMBean { if (r.contains(sstableRange)) { - logger.info("[repair #{}] SSTable {} fully contained in range {}, mutating repairedAt instead of anticompacting", parentRepairSession, sstable, r); + logger.info("{} SSTable {} fully contained in range {}, mutating repairedAt instead of anticompacting", PreviewKind.NONE.logPrefix(parentRepairSession), sstable, r); sstable.descriptor.getMetadataSerializer().mutateRepaired(sstable.descriptor, repairedAt, pendingRepair); sstable.reloadSSTableMetadata(); mutatedRepairStatuses.add(sstable); @@ -652,14 +657,14 @@ public class CompactionManager implements CompactionManagerMBean } else if (sstableRange.intersects(r)) { - logger.info("[repair #{}] SSTable {} ({}) will be anticompacted on range {}", parentRepairSession, sstable, sstableRange, r); + logger.info("{} SSTable {} ({}) will be anticompacted on range {}", PreviewKind.NONE.logPrefix(parentRepairSession), sstable, sstableRange, r); shouldAnticompact = true; } } if (!shouldAnticompact) { - logger.info("[repair #{}] SSTable {} ({}) does not intersect repaired ranges {}, not touching repairedAt.", parentRepairSession, sstable, sstableRange, normalizedRanges); + logger.info("{} SSTable {} ({}) does not intersect repaired ranges {}, not touching repairedAt.", PreviewKind.NONE.logPrefix(parentRepairSession), sstable, sstableRange, normalizedRanges); nonAnticompacting.add(sstable); sstableIterator.remove(); } @@ -678,7 +683,7 @@ public class CompactionManager implements CompactionManagerMBean txn.close(); } - logger.info("[repair #{}] Completed anticompaction successfully", parentRepairSession); + logger.info("{} Completed anticompaction successfully", PreviewKind.NONE.logPrefix(parentRepairSession)); } public void performMaximal(final ColumnFamilyStore cfStore, boolean splitOutput) @@ -1427,7 +1432,12 @@ public class CompactionManager implements CompactionManagerMBean Set sstablesToValidate = new HashSet<>(); com.google.common.base.Predicate predicate; - if (validator.isConsistent) + if (prs.isPreview()) + { + predicate = prs.getPreviewPredicate(); + + } + else if (validator.isConsistent) { predicate = s -> validator.desc.parentSessionId.equals(s.getSSTableMetadata().pendingRepair); } diff --git a/src/java/org/apache/cassandra/dht/RangeStreamer.java b/src/java/org/apache/cassandra/dht/RangeStreamer.java index fd976c9510..85c2307e8e 100644 --- a/src/java/org/apache/cassandra/dht/RangeStreamer.java +++ b/src/java/org/apache/cassandra/dht/RangeStreamer.java @@ -39,6 +39,7 @@ import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.IEndpointSnitch; import org.apache.cassandra.locator.TokenMetadata; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.streaming.StreamPlan; import org.apache.cassandra.streaming.StreamResultFuture; import org.apache.cassandra.streaming.StreamOperation; @@ -156,7 +157,7 @@ public class RangeStreamer this.tokens = tokens; this.address = address; this.description = streamOperation.getDescription(); - this.streamPlan = new StreamPlan(streamOperation, connectionsPerHost, true, connectSequentially, null); + this.streamPlan = new StreamPlan(streamOperation, connectionsPerHost, true, connectSequentially, null, PreviewKind.NONE); this.useStrictConsistency = useStrictConsistency; this.snitch = snitch; this.stateStore = stateStore; diff --git a/src/java/org/apache/cassandra/exceptions/RepairException.java b/src/java/org/apache/cassandra/exceptions/RepairException.java index 2f5f2c1c72..db219a21ec 100644 --- a/src/java/org/apache/cassandra/exceptions/RepairException.java +++ b/src/java/org/apache/cassandra/exceptions/RepairException.java @@ -18,6 +18,7 @@ package org.apache.cassandra.exceptions; import org.apache.cassandra.repair.RepairJobDesc; +import org.apache.cassandra.streaming.PreviewKind; /** * Exception thrown during repair @@ -25,22 +26,23 @@ import org.apache.cassandra.repair.RepairJobDesc; public class RepairException extends Exception { public final RepairJobDesc desc; + public final PreviewKind previewKind; public RepairException(RepairJobDesc desc, String message) { - super(message); - this.desc = desc; + this(desc, null, message); } - public RepairException(RepairJobDesc desc, String message, Throwable cause) + public RepairException(RepairJobDesc desc, PreviewKind previewKind, String message) { - super(message, cause); + super(message); this.desc = desc; + this.previewKind = previewKind != null ? previewKind : PreviewKind.NONE; } @Override public String getMessage() { - return desc + " " + super.getMessage(); + return desc.toString(previewKind) + ' ' + super.getMessage(); } } diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java index e9ea35a9b5..ff47becccb 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java @@ -159,7 +159,7 @@ public class SSTableLoader implements StreamEventHandler client.init(keyspace); outputHandler.output("Established connection to initial hosts"); - StreamPlan plan = new StreamPlan(StreamOperation.BULK_LOAD, connectionsPerHost, false, false, null).connectionFactory(client.getConnectionFactory()); + StreamPlan plan = new StreamPlan(StreamOperation.BULK_LOAD, connectionsPerHost, false, false, null, PreviewKind.NONE).connectionFactory(client.getConnectionFactory()); Map>> endpointToRanges = client.getEndpointToRangesMap(); openSSTables(endpointToRanges); diff --git a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java index e5fdc99c22..8db5fcbe6f 100644 --- a/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java +++ b/src/java/org/apache/cassandra/net/IncomingStreamingConnection.java @@ -73,7 +73,7 @@ public class IncomingStreamingConnection extends Thread implements Closeable // The receiving side distinguish two connections by looking at StreamInitMessage#isForOutgoing. // Note: we cannot use the same socket for incoming and outgoing streams because we want to // parallelize said streams and the socket is blocking, so we might deadlock. - StreamResultFuture.initReceivingSide(init.sessionIndex, init.planId, init.streamOperation, init.from, this, init.isForOutgoing, version, init.keepSSTableLevel, init.pendingRepair); + StreamResultFuture.initReceivingSide(init.sessionIndex, init.planId, init.streamOperation, init.from, this, init.isForOutgoing, version, init.keepSSTableLevel, init.pendingRepair, init.previewKind); } catch (Throwable t) { diff --git a/src/java/org/apache/cassandra/repair/LocalSyncTask.java b/src/java/org/apache/cassandra/repair/LocalSyncTask.java index 06b16612ce..5bb66dcc39 100644 --- a/src/java/org/apache/cassandra/repair/LocalSyncTask.java +++ b/src/java/org/apache/cassandra/repair/LocalSyncTask.java @@ -29,6 +29,7 @@ import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.streaming.ProgressInfo; import org.apache.cassandra.streaming.StreamEvent; import org.apache.cassandra.streaming.StreamEventHandler; @@ -38,6 +39,8 @@ import org.apache.cassandra.streaming.StreamOperation; import org.apache.cassandra.tracing.TraceState; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.MerkleTree; +import org.apache.cassandra.utils.MerkleTrees; /** * LocalSyncTask performs streaming between local(coordinator) node and remote replica. @@ -51,9 +54,9 @@ public class LocalSyncTask extends SyncTask implements StreamEventHandler private final UUID pendingRepair; private final boolean pullRepair; - public LocalSyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2, UUID pendingRepair, boolean pullRepair) + public LocalSyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2, UUID pendingRepair, boolean pullRepair, PreviewKind previewKind) { - super(desc, r1, r2); + super(desc, r1, r2, previewKind); this.pendingRepair = pendingRepair; this.pullRepair = pullRepair; } @@ -62,7 +65,7 @@ public class LocalSyncTask extends SyncTask implements StreamEventHandler @VisibleForTesting StreamPlan createStreamPlan(InetAddress dst, InetAddress preferred, List> differences) { - StreamPlan plan = new StreamPlan(StreamOperation.REPAIR, 1, false, false, pendingRepair) + StreamPlan plan = new StreamPlan(StreamOperation.REPAIR, 1, false, false, pendingRepair, previewKind) .listeners(this) .flushBeforeTransfer(pendingRepair == null) .requestRanges(dst, preferred, desc.keyspace, differences, desc.columnFamily); // request ranges from the remote node @@ -79,6 +82,7 @@ public class LocalSyncTask extends SyncTask implements StreamEventHandler * 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. */ + @Override protected void startSync(List> differences) { InetAddress local = FBUtilities.getBroadcastAddress(); @@ -87,7 +91,7 @@ public class LocalSyncTask extends SyncTask implements StreamEventHandler InetAddress preferred = SystemKeyspace.getPreferredIP(dst); String message = String.format("Performing streaming repair of %d ranges with %s", differences.size(), dst); - logger.info("[repair #{}] {}", desc.sessionId, message); + logger.info("{} {}", previewKind.logPrefix(desc.sessionId), message); Tracing.traceRepair(message); createStreamPlan(dst, preferred, differences).execute(); @@ -122,9 +126,9 @@ public class LocalSyncTask extends SyncTask implements StreamEventHandler public void onSuccess(StreamState result) { String message = String.format("Sync complete using session %s between %s and %s on %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily); - logger.info("[repair #{}] {}", desc.sessionId, message); + logger.info("{} {}", previewKind.logPrefix(desc.sessionId), message); Tracing.traceRepair(message); - set(stat); + set(stat.withSummaries(result.createSummaries())); } public void onFailure(Throwable t) diff --git a/src/java/org/apache/cassandra/repair/RemoteSyncTask.java b/src/java/org/apache/cassandra/repair/RemoteSyncTask.java index ededc40700..a26a3923df 100644 --- a/src/java/org/apache/cassandra/repair/RemoteSyncTask.java +++ b/src/java/org/apache/cassandra/repair/RemoteSyncTask.java @@ -28,6 +28,8 @@ 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.streaming.PreviewKind; +import org.apache.cassandra.streaming.SessionSummary; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.FBUtilities; @@ -41,30 +43,31 @@ public class RemoteSyncTask extends SyncTask { private static final Logger logger = LoggerFactory.getLogger(RemoteSyncTask.class); - public RemoteSyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2) + public RemoteSyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2, PreviewKind previewKind) { - super(desc, r1, r2); + super(desc, r1, r2, previewKind); } + @Override protected void startSync(List> differences) { InetAddress local = FBUtilities.getBroadcastAddress(); - SyncRequest request = new SyncRequest(desc, local, r1.endpoint, r2.endpoint, differences); + SyncRequest request = new SyncRequest(desc, local, r1.endpoint, r2.endpoint, differences, previewKind); String message = String.format("Forwarding streaming repair of %d ranges to %s (to be streamed with %s)", request.ranges.size(), request.src, request.dst); - logger.info("[repair #{}] {}", desc.sessionId, message); + logger.info("{} {}", previewKind.logPrefix(desc.sessionId), message); Tracing.traceRepair(message); MessagingService.instance().sendOneWay(request.createMessage(), request.src); } - public void syncComplete(boolean success) + public void syncComplete(boolean success, List summaries) { if (success) { - set(stat); + set(stat.withSummaries(summaries)); } else { - setException(new RepairException(desc, String.format("Sync failed between %s and %s", r1.endpoint, r2.endpoint))); + setException(new RepairException(desc, previewKind, 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 d1f6a94dbf..d6c1176907 100644 --- a/src/java/org/apache/cassandra/repair/RepairJob.java +++ b/src/java/org/apache/cassandra/repair/RepairJob.java @@ -26,6 +26,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; @@ -42,6 +43,7 @@ public class RepairJob extends AbstractFuture implements Runnable private final RepairParallelism parallelismDegree; private final ListeningExecutorService taskExecutor; private final boolean isConsistent; + private final PreviewKind previewKind; /** * Create repair job to run on specific columnfamily @@ -49,13 +51,14 @@ public class RepairJob extends AbstractFuture implements Runnable * @param session RepairSession that this RepairJob belongs * @param columnFamily name of the ColumnFamily to repair */ - public RepairJob(RepairSession session, String columnFamily, boolean isConsistent) + public RepairJob(RepairSession session, String columnFamily, boolean isConsistent, PreviewKind previewKind) { this.session = session; this.desc = new RepairJobDesc(session.parentRepairSession, session.getId(), session.keyspace, columnFamily, session.getRanges()); this.taskExecutor = session.taskExecutor; this.parallelismDegree = session.parallelismDegree; this.isConsistent = isConsistent; + this.previewKind = previewKind; } /** @@ -128,11 +131,11 @@ public class RepairJob extends AbstractFuture implements Runnable SyncTask task; if (r1.endpoint.equals(local) || r2.endpoint.equals(local)) { - task = new LocalSyncTask(desc, r1, r2, isConsistent ? desc.parentSessionId : null, session.pullRepair); + task = new LocalSyncTask(desc, r1, r2, isConsistent ? desc.parentSessionId : null, session.pullRepair, session.previewKind); } else { - task = new RemoteSyncTask(desc, r1, r2); + task = new RemoteSyncTask(desc, r1, r2, session.previewKind); // 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); @@ -150,8 +153,11 @@ public class RepairJob extends AbstractFuture implements Runnable { public void onSuccess(List stats) { - logger.info("[repair #{}] {} is fully synced", session.getId(), desc.columnFamily); - SystemDistributedKeyspace.successfulRepairJob(session.getId(), desc.keyspace, desc.columnFamily); + if (!previewKind.isPreview()) + { + logger.info("{} {} is fully synced", previewKind.logPrefix(session.getId()), desc.columnFamily); + SystemDistributedKeyspace.successfulRepairJob(session.getId(), desc.keyspace, desc.columnFamily); + } set(new RepairResult(desc, stats)); } @@ -160,8 +166,11 @@ public class RepairJob extends AbstractFuture implements Runnable */ public void onFailure(Throwable t) { - logger.warn("[repair #{}] {} sync failed", session.getId(), desc.columnFamily); - SystemDistributedKeyspace.failedRepairJob(session.getId(), desc.keyspace, desc.columnFamily, t); + if (!previewKind.isPreview()) + { + logger.warn("{} {} sync failed", previewKind.logPrefix(session.getId()), desc.columnFamily); + SystemDistributedKeyspace.failedRepairJob(session.getId(), desc.keyspace, desc.columnFamily, t); + } setException(t); } }, taskExecutor); @@ -179,13 +188,13 @@ public class RepairJob extends AbstractFuture implements Runnable private ListenableFuture> sendValidationRequest(Collection endpoints) { String message = String.format("Requesting merkle trees for %s (to %s)", desc.columnFamily, endpoints); - logger.info("[repair #{}] {}", desc.sessionId, message); + logger.info("{} {}", previewKind.logPrefix(desc.sessionId), message); Tracing.traceRepair(message); int gcBefore = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily).gcBefore(FBUtilities.nowInSeconds()); List> tasks = new ArrayList<>(endpoints.size()); for (InetAddress endpoint : endpoints) { - ValidationTask task = new ValidationTask(desc, endpoint, gcBefore); + ValidationTask task = new ValidationTask(desc, endpoint, gcBefore, previewKind); tasks.add(task); session.waitForValidation(Pair.create(desc, endpoint), task); taskExecutor.execute(task); @@ -199,14 +208,14 @@ public class RepairJob extends AbstractFuture implements Runnable private ListenableFuture> sendSequentialValidationRequest(Collection endpoints) { String message = String.format("Requesting merkle trees for %s (to %s)", desc.columnFamily, endpoints); - logger.info("[repair #{}] {}", desc.sessionId, message); + logger.info("{} {}", previewKind.logPrefix(desc.sessionId), message); Tracing.traceRepair(message); int gcBefore = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily).gcBefore(FBUtilities.nowInSeconds()); List> tasks = new ArrayList<>(endpoints.size()); Queue requests = new LinkedList<>(endpoints); InetAddress address = requests.poll(); - ValidationTask firstTask = new ValidationTask(desc, address, gcBefore); + ValidationTask firstTask = new ValidationTask(desc, address, gcBefore, previewKind); logger.info("Validating {}", address); session.waitForValidation(Pair.create(desc, address), firstTask); tasks.add(firstTask); @@ -214,7 +223,7 @@ public class RepairJob extends AbstractFuture implements Runnable while (requests.size() > 0) { final InetAddress nextAddress = requests.poll(); - final ValidationTask nextTask = new ValidationTask(desc, nextAddress, gcBefore); + final ValidationTask nextTask = new ValidationTask(desc, nextAddress, gcBefore, previewKind); tasks.add(nextTask); Futures.addCallback(currentTask, new FutureCallback() { @@ -241,7 +250,7 @@ public class RepairJob extends AbstractFuture implements Runnable private ListenableFuture> sendDCAwareValidationRequest(Collection endpoints) { String message = String.format("Requesting merkle trees for %s (to %s)", desc.columnFamily, endpoints); - logger.info("[repair #{}] {}", desc.sessionId, message); + logger.info("{} {}", previewKind.logPrefix(desc.sessionId), message); Tracing.traceRepair(message); int gcBefore = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily).gcBefore(FBUtilities.nowInSeconds()); List> tasks = new ArrayList<>(endpoints.size()); @@ -263,7 +272,7 @@ public class RepairJob extends AbstractFuture implements Runnable { Queue requests = entry.getValue(); InetAddress address = requests.poll(); - ValidationTask firstTask = new ValidationTask(desc, address, gcBefore); + ValidationTask firstTask = new ValidationTask(desc, address, gcBefore, previewKind); logger.info("Validating {}", address); session.waitForValidation(Pair.create(desc, address), firstTask); tasks.add(firstTask); @@ -271,7 +280,7 @@ public class RepairJob extends AbstractFuture implements Runnable while (requests.size() > 0) { final InetAddress nextAddress = requests.poll(); - final ValidationTask nextTask = new ValidationTask(desc, nextAddress, gcBefore); + final ValidationTask nextTask = new ValidationTask(desc, nextAddress, gcBefore, previewKind); tasks.add(nextTask); Futures.addCallback(currentTask, new FutureCallback() { diff --git a/src/java/org/apache/cassandra/repair/RepairJobDesc.java b/src/java/org/apache/cassandra/repair/RepairJobDesc.java index be3daef804..8f70e005c0 100644 --- a/src/java/org/apache/cassandra/repair/RepairJobDesc.java +++ b/src/java/org/apache/cassandra/repair/RepairJobDesc.java @@ -32,6 +32,7 @@ import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.utils.UUIDSerializer; /** @@ -66,6 +67,11 @@ public class RepairJobDesc return "[repair #" + sessionId + " on " + keyspace + "/" + columnFamily + ", " + ranges + "]"; } + public String toString(PreviewKind previewKind) + { + return '[' + previewKind.logPrefix() + " #" + sessionId + " on " + keyspace + "/" + columnFamily + ", " + ranges + "]"; + } + @Override public boolean equals(Object o) { diff --git a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java index b6b9f87e15..ed62229e6a 100644 --- a/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java +++ b/src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java @@ -35,6 +35,7 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.repair.messages.*; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.streaming.PreviewKind; /** * Handles all repair related message. @@ -50,6 +51,12 @@ public class RepairMessageVerbHandler implements IVerbHandler return ActiveRepairService.instance.consistent.local.isSessionInProgress(sessionID); } + private PreviewKind previewKind(UUID sessionID) + { + ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(sessionID); + return prs != null ? prs.previewKind : PreviewKind.NONE; + } + public void doVerb(final MessageIn message, final int id) { // TODO add cancel/interrupt message @@ -79,7 +86,8 @@ public class RepairMessageVerbHandler implements IVerbHandler prepareMessage.ranges, prepareMessage.isIncremental, prepareMessage.timestamp, - prepareMessage.isGlobal); + prepareMessage.isGlobal, + prepareMessage.previewKind); MessagingService.instance().sendReply(new MessageOut(MessagingService.Verb.INTERNAL_RESPONSE), id, message.from); break; @@ -127,7 +135,8 @@ public class RepairMessageVerbHandler implements IVerbHandler } ActiveRepairService.instance.consistent.local.maybeSetRepairing(desc.parentSessionId); - Validator validator = new Validator(desc, message.from, validationRequest.gcBefore, isConsistent(desc.parentSessionId)); + Validator validator = new Validator(desc, message.from, validationRequest.gcBefore, + isConsistent(desc.parentSessionId), previewKind(desc.parentSessionId)); CompactionManager.instance.submitValidation(store, validator); break; @@ -135,7 +144,7 @@ public class RepairMessageVerbHandler implements IVerbHandler // forwarded sync request SyncRequest request = (SyncRequest) message.payload; logger.debug("Syncing {}", request); - StreamingRepairTask task = new StreamingRepairTask(desc, request, isConsistent(desc.parentSessionId) ? desc.parentSessionId : null); + StreamingRepairTask task = new StreamingRepairTask(desc, request, isConsistent(desc.parentSessionId) ? desc.parentSessionId : null, request.previewKind); task.run(); break; diff --git a/src/java/org/apache/cassandra/repair/RepairRunnable.java b/src/java/org/apache/cassandra/repair/RepairRunnable.java index b8bef95cd5..354d8b5b17 100644 --- a/src/java/org/apache/cassandra/repair/RepairRunnable.java +++ b/src/java/org/apache/cassandra/repair/RepairRunnable.java @@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.JMXConfigurableThreadPoolExecutor; import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.repair.consistent.SyncStatSummary; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.cql3.QueryProcessor; @@ -50,6 +51,7 @@ import org.apache.cassandra.repair.messages.RepairOption; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.service.QueryState; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.tracing.TraceKeyspace; import org.apache.cassandra.tracing.TraceState; import org.apache.cassandra.tracing.Tracing; @@ -218,7 +220,11 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti cfnames[i] = columnFamilyStores.get(i).name; } - SystemDistributedKeyspace.startParentRepair(parentSession, keyspace, cfnames, options); + if (!options.isPreview()) + { + SystemDistributedKeyspace.startParentRepair(parentSession, keyspace, cfnames, options); + } + long repairedAt; try { @@ -228,12 +234,19 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti } catch (Throwable t) { - SystemDistributedKeyspace.failParentRepair(parentSession, t); + if (!options.isPreview()) + { + SystemDistributedKeyspace.failParentRepair(parentSession, t); + } fireErrorAndComplete(tag, progress.get(), totalProgress, t.getMessage()); return; } - if (options.isIncremental()) + if (options.isPreview()) + { + previewRepair(parentSession, repairedAt, startTime, traceState, allNeighbors, commonRanges, cfnames); + } + else if (options.isIncremental()) { consistentRepair(parentSession, repairedAt, startTime, traceState, allNeighbors, commonRanges, cfnames); } @@ -311,6 +324,76 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti Futures.addCallback(repairResult, new RepairCompleteCallback(parentSession, ranges, startTime, traceState, hasFailure, executor)); } + private void previewRepair(UUID parentSession, + long repairedAt, + long startTime, + TraceState traceState, + Set allNeighbors, + List, ? extends Collection>>> commonRanges, + String... cfnames) + { + + logger.debug("Starting preview repair for {}", parentSession); + // Set up RepairJob executor for this repair command. + ListeningExecutorService executor = createExecutor(); + + final ListenableFuture> allSessions = submitRepairSessions(parentSession, false, executor, commonRanges, cfnames); + + Futures.addCallback(allSessions, new FutureCallback>() + { + public void onSuccess(List results) + { + try + { + PreviewKind previewKind = options.getPreviewKind(); + assert previewKind != PreviewKind.NONE; + SyncStatSummary summary = new SyncStatSummary(true); + summary.consumeSessionResults(results); + + if (summary.isEmpty()) + { + String message = previewKind == PreviewKind.REPAIRED ? "Repaired data is in sync" : "Previewed data was in sync"; + logger.info(message); + fireProgressEvent(tag, new ProgressEvent(ProgressEventType.NOTIFICATION, progress.get(), totalProgress, message)); + } + else + { + String message = (previewKind == PreviewKind.REPAIRED ? "Repaired data is inconsistent\n" : "Preview complete\n") + summary.toString(); + logger.info(message); + fireProgressEvent(tag, new ProgressEvent(ProgressEventType.NOTIFICATION, progress.get(), totalProgress, message)); + } + + fireProgressEvent(tag, new ProgressEvent(ProgressEventType.SUCCESS, progress.get(), totalProgress, + "Repair preview completed successfully")); + complete(); + } + catch (Throwable t) + { + logger.error("Error completing preview repair", t); + onFailure(t); + } + } + + public void onFailure(Throwable t) + { + fireProgressEvent(tag, new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, t.getMessage())); + logger.error("Error completing preview repair", t); + complete(); + } + + private void 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(tag, new ProgressEvent(ProgressEventType.COMPLETE, progress.get(), totalProgress, message)); + executor.shutdownNow(); + } + }); + } + private ListenableFuture> submitRepairSessions(UUID parentSession, boolean isConsistent, ListeningExecutorService executor, @@ -327,6 +410,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti p.left, isConsistent, options.isPullRepair(), + options.getPreviewKind(), executor, cfnames); if (session == null) @@ -405,7 +489,10 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti public void onSuccess(Object result) { - SystemDistributedKeyspace.successfulParentRepair(parentSession, successfulRanges); + if (!options.isPreview()) + { + SystemDistributedKeyspace.successfulParentRepair(parentSession, successfulRanges); + } if (hasFailure.get()) { fireProgressEvent(tag, new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, @@ -422,7 +509,10 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti public void onFailure(Throwable t) { fireProgressEvent(tag, new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, t.getMessage())); - SystemDistributedKeyspace.failParentRepair(parentSession, t); + if (!options.isPreview()) + { + SystemDistributedKeyspace.failParentRepair(parentSession, t); + } repairComplete(); } diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java index 2aa068c87f..8aa95903c6 100644 --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@ -34,6 +34,8 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.gms.*; +import org.apache.cassandra.streaming.PreviewKind; +import org.apache.cassandra.streaming.SessionSummary; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MerkleTrees; @@ -91,6 +93,7 @@ public class RepairSession extends AbstractFuture implement public final Collection> ranges; public final Set endpoints; public final boolean isConsistent; + public final PreviewKind previewKind; private final AtomicBoolean isFailed = new AtomicBoolean(false); @@ -125,6 +128,7 @@ public class RepairSession extends AbstractFuture implement Set endpoints, boolean isConsistent, boolean pullRepair, + PreviewKind previewKind, String... cfnames) { assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it"; @@ -137,6 +141,7 @@ public class RepairSession extends AbstractFuture implement this.ranges = ranges; this.endpoints = endpoints; this.isConsistent = isConsistent; + this.previewKind = previewKind; this.pullRepair = pullRepair; } @@ -177,7 +182,7 @@ public class RepairSession extends AbstractFuture implement } String message = String.format("Received merkle tree for %s from %s", desc.columnFamily, endpoint); - logger.info("[repair #{}] {}", getId(), message); + logger.info("{} {}", previewKind.logPrefix(getId()), message); Tracing.traceRepair(message); task.treesReceived(trees); } @@ -189,7 +194,7 @@ public class RepairSession extends AbstractFuture implement * @param nodes nodes that completed sync * @param success true if sync succeeded */ - public void syncComplete(RepairJobDesc desc, NodePair nodes, boolean success) + public void syncComplete(RepairJobDesc desc, NodePair nodes, boolean success, List summaries) { RemoteSyncTask task = syncingTasks.get(Pair.create(desc, nodes)); if (task == null) @@ -198,8 +203,8 @@ public class RepairSession extends AbstractFuture implement return; } - logger.debug("[repair #{}] Repair completed between {} and {} on {}", getId(), nodes.endpoint1, nodes.endpoint2, desc.columnFamily); - task.syncComplete(success); + logger.debug("{} Repair completed between {} and {} on {}", previewKind.logPrefix(getId()), nodes.endpoint1, nodes.endpoint2, desc.columnFamily); + task.syncComplete(success, summaries); } private String repairedNodes() @@ -225,16 +230,22 @@ public class RepairSession extends AbstractFuture implement if (terminated) return; - logger.info("[repair #{}] new session: will sync {} on range {} for {}.{}", getId(), repairedNodes(), ranges, keyspace, Arrays.toString(cfnames)); + logger.info("{} new session: will sync {} on range {} for {}.{}", previewKind.logPrefix(getId()), repairedNodes(), ranges, keyspace, Arrays.toString(cfnames)); Tracing.traceRepair("Syncing range {}", ranges); - SystemDistributedKeyspace.startRepairs(getId(), parentRepairSession, keyspace, cfnames, ranges, endpoints); + if (!previewKind.isPreview()) + { + SystemDistributedKeyspace.startRepairs(getId(), parentRepairSession, keyspace, cfnames, ranges, endpoints); + } if (endpoints.isEmpty()) { - logger.info("[repair #{}] {}", getId(), message = String.format("No neighbors to repair with on range %s: session completed", ranges)); + logger.info("{} {}", previewKind.logPrefix(getId()), message = String.format("No neighbors to repair with on range %s: session completed", ranges)); Tracing.traceRepair(message); set(new RepairSessionResult(id, keyspace, ranges, Lists.newArrayList())); - SystemDistributedKeyspace.failRepairs(getId(), keyspace, cfnames, new RuntimeException(message)); + if (!previewKind.isPreview()) + { + SystemDistributedKeyspace.failRepairs(getId(), keyspace, cfnames, new RuntimeException(message)); + } return; } @@ -244,10 +255,13 @@ public class RepairSession extends AbstractFuture implement if (!FailureDetector.instance.isAlive(endpoint)) { message = String.format("Cannot proceed on repair because a neighbor (%s) is dead: session failed", endpoint); - logger.error("[repair #{}] {}", getId(), message); + logger.error("{} {}", previewKind.logPrefix(getId()), message); Exception e = new IOException(message); setException(e); - SystemDistributedKeyspace.failRepairs(getId(), keyspace, cfnames, e); + if (!previewKind.isPreview()) + { + SystemDistributedKeyspace.failRepairs(getId(), keyspace, cfnames, e); + } return; } } @@ -256,7 +270,7 @@ public class RepairSession extends AbstractFuture implement List> jobs = new ArrayList<>(cfnames.length); for (String cfname : cfnames) { - RepairJob job = new RepairJob(this, cfname, isConsistent); + RepairJob job = new RepairJob(this, cfname, isConsistent, previewKind); executor.execute(job); jobs.add(job); } @@ -267,7 +281,7 @@ public class RepairSession extends AbstractFuture implement public void onSuccess(List results) { // this repair session is completed - logger.info("[repair #{}] {}", getId(), "Session completed successfully"); + logger.info("{} {}", previewKind.logPrefix(getId()), "Session completed successfully"); Tracing.traceRepair("Completed sync of range {}", ranges); set(new RepairSessionResult(id, keyspace, ranges, results)); @@ -278,7 +292,7 @@ public class RepairSession extends AbstractFuture implement public void onFailure(Throwable t) { - logger.error(String.format("[repair #%s] Session completed with the following error", getId()), t); + logger.error("{} Session completed with the following error", previewKind.logPrefix(getId()), t); Tracing.traceRepair("Session completed with the following error: {}", t); forceShutdown(t); } @@ -335,7 +349,7 @@ public class RepairSession extends AbstractFuture implement return; Exception exception = new IOException(String.format("Endpoint %s died", endpoint)); - logger.error(String.format("[repair #%s] session completed with the following error", getId()), exception); + logger.error(String.format("{} session completed with the following error", previewKind.logPrefix(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/StreamingRepairTask.java b/src/java/org/apache/cassandra/repair/StreamingRepairTask.java index 7042de1b3d..f43010b248 100644 --- a/src/java/org/apache/cassandra/repair/StreamingRepairTask.java +++ b/src/java/org/apache/cassandra/repair/StreamingRepairTask.java @@ -19,6 +19,7 @@ package org.apache.cassandra.repair; import java.net.InetAddress; import java.util.UUID; +import java.util.Collections; import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; @@ -28,7 +29,7 @@ import org.apache.cassandra.db.SystemKeyspace; 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.PreviewKind; import org.apache.cassandra.streaming.StreamEvent; import org.apache.cassandra.streaming.StreamEventHandler; import org.apache.cassandra.streaming.StreamPlan; @@ -46,12 +47,14 @@ public class StreamingRepairTask implements Runnable, StreamEventHandler private final RepairJobDesc desc; private final SyncRequest request; private final UUID pendingRepair; + private final PreviewKind previewKind; - public StreamingRepairTask(RepairJobDesc desc, SyncRequest request, UUID pendingRepair) + public StreamingRepairTask(RepairJobDesc desc, SyncRequest request, UUID pendingRepair, PreviewKind previewKind) { this.desc = desc; this.request = request; this.pendingRepair = pendingRepair; + this.previewKind = previewKind; } public void run() @@ -65,7 +68,7 @@ public class StreamingRepairTask implements Runnable, StreamEventHandler @VisibleForTesting StreamPlan createStreamPlan(InetAddress dest, InetAddress preferred) { - return new StreamPlan(StreamOperation.REPAIR, 1, false, false, pendingRepair) + return new StreamPlan(StreamOperation.REPAIR, 1, false, false, pendingRepair, previewKind) .listeners(this) .flushBeforeTransfer(pendingRepair == null) // sstables are isolated at the beginning of an incremental repair session, so flushing isn't neccessary .requestRanges(dest, preferred, desc.keyspace, request.ranges, desc.columnFamily) // request ranges from the remote node @@ -83,8 +86,8 @@ public class StreamingRepairTask implements Runnable, StreamEventHandler */ public void onSuccess(StreamState state) { - logger.info("[repair #{}] streaming task succeed, returning response to {}", desc.sessionId, request.initiator); - MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, true).createMessage(), request.initiator); + logger.info("{} streaming task succeed, returning response to {}", previewKind.logPrefix(desc.sessionId), request.initiator); + MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, true, state.createSummaries()).createMessage(), request.initiator); } /** @@ -92,6 +95,6 @@ public class StreamingRepairTask implements Runnable, StreamEventHandler */ public void onFailure(Throwable t) { - MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, false).createMessage(), request.initiator); + MessagingService.instance().sendOneWay(new SyncComplete(desc, request.src, request.dst, false, Collections.emptyList()).createMessage(), request.initiator); } } diff --git a/src/java/org/apache/cassandra/repair/SyncStat.java b/src/java/org/apache/cassandra/repair/SyncStat.java index 5721a2083b..dab56597ca 100644 --- a/src/java/org/apache/cassandra/repair/SyncStat.java +++ b/src/java/org/apache/cassandra/repair/SyncStat.java @@ -17,17 +17,33 @@ */ package org.apache.cassandra.repair; +import java.util.List; + +import org.apache.cassandra.streaming.SessionSummary; + /** * Statistics about synchronizing two replica */ public class SyncStat { public final NodePair nodes; - public final long numberOfDifferences; + public final long numberOfDifferences; // TODO: revert to Range + public final List summaries; public SyncStat(NodePair nodes, long numberOfDifferences) + { + this(nodes, numberOfDifferences, null); + } + + public SyncStat(NodePair nodes, long numberOfDifferences, List summaries) { this.nodes = nodes; this.numberOfDifferences = numberOfDifferences; + this.summaries = summaries; + } + + public SyncStat withSummaries(List summaries) + { + return new SyncStat(nodes, numberOfDifferences, summaries); } } diff --git a/src/java/org/apache/cassandra/repair/SyncTask.java b/src/java/org/apache/cassandra/repair/SyncTask.java index 8adec6f4a0..6f3157a99a 100644 --- a/src/java/org/apache/cassandra/repair/SyncTask.java +++ b/src/java/org/apache/cassandra/repair/SyncTask.java @@ -25,6 +25,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.MerkleTrees; @@ -39,14 +40,16 @@ public abstract class SyncTask extends AbstractFuture implements Runna protected final RepairJobDesc desc; protected final TreeResponse r1; protected final TreeResponse r2; + protected final PreviewKind previewKind; protected volatile SyncStat stat; - public SyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2) + public SyncTask(RepairJobDesc desc, TreeResponse r1, TreeResponse r2, PreviewKind previewKind) { this.desc = desc; this.r1 = r1; this.r2 = r2; + this.previewKind = previewKind; } /** @@ -60,7 +63,7 @@ public abstract class SyncTask extends AbstractFuture implements Runna 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); + String format = String.format("%s Endpoints %s and %s %%s for %s", previewKind.logPrefix(desc.sessionId), r1.endpoint, r2.endpoint, desc.columnFamily); if (differences.isEmpty()) { logger.info(String.format(format, "are consistent")); diff --git a/src/java/org/apache/cassandra/repair/ValidationTask.java b/src/java/org/apache/cassandra/repair/ValidationTask.java index 177ad3edae..f68d3c5a79 100644 --- a/src/java/org/apache/cassandra/repair/ValidationTask.java +++ b/src/java/org/apache/cassandra/repair/ValidationTask.java @@ -24,6 +24,7 @@ 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.streaming.PreviewKind; import org.apache.cassandra.utils.MerkleTrees; /** @@ -35,12 +36,14 @@ public class ValidationTask extends AbstractFuture implements Runn private final RepairJobDesc desc; private final InetAddress endpoint; private final int gcBefore; + private final PreviewKind previewKind; - public ValidationTask(RepairJobDesc desc, InetAddress endpoint, int gcBefore) + public ValidationTask(RepairJobDesc desc, InetAddress endpoint, int gcBefore, PreviewKind previewKind) { this.desc = desc; this.endpoint = endpoint; this.gcBefore = gcBefore; + this.previewKind = previewKind; } /** @@ -61,7 +64,7 @@ public class ValidationTask extends AbstractFuture implements Runn { if (trees == null) { - setException(new RepairException(desc, "Validation failed in " + endpoint)); + setException(new RepairException(desc, previewKind, "Validation failed in " + endpoint)); } else { diff --git a/src/java/org/apache/cassandra/repair/Validator.java b/src/java/org/apache/cassandra/repair/Validator.java index e8e3621a57..ba1fa9dba0 100644 --- a/src/java/org/apache/cassandra/repair/Validator.java +++ b/src/java/org/apache/cassandra/repair/Validator.java @@ -37,6 +37,7 @@ import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.repair.messages.ValidationComplete; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MerkleTree; @@ -71,22 +72,25 @@ public class Validator implements Runnable // last key seen private DecoratedKey lastKey; - public Validator(RepairJobDesc desc, InetAddress initiator, int gcBefore) + private final PreviewKind previewKind; + + public Validator(RepairJobDesc desc, InetAddress initiator, int gcBefore, PreviewKind previewKind) { - this(desc, initiator, gcBefore, false, false); + this(desc, initiator, gcBefore, false, false, previewKind); } - public Validator(RepairJobDesc desc, InetAddress initiator, int gcBefore, boolean isConsistent) + public Validator(RepairJobDesc desc, InetAddress initiator, int gcBefore, boolean isConsistent, PreviewKind previewKind) { - this(desc, initiator, gcBefore, false, isConsistent); + this(desc, initiator, gcBefore, false, isConsistent, previewKind); } - public Validator(RepairJobDesc desc, InetAddress initiator, int gcBefore, boolean evenTreeDistribution, boolean isConsistent) + public Validator(RepairJobDesc desc, InetAddress initiator, int gcBefore, boolean evenTreeDistribution, boolean isConsistent, PreviewKind previewKind) { this.desc = desc; this.initiator = initiator; this.gcBefore = gcBefore; this.isConsistent = isConsistent; + this.previewKind = previewKind; validated = 0; range = null; ranges = null; @@ -285,7 +289,7 @@ public class Validator implements Runnable // respond to the request that triggered this validation if (!initiator.equals(FBUtilities.getBroadcastAddress())) { - logger.info("[repair #{}] Sending completed merkle tree to {} for {}.{}", desc.sessionId, initiator, desc.keyspace, desc.columnFamily); + logger.info("{} Sending completed merkle tree to {} for {}.{}", previewKind.logPrefix(desc.sessionId), initiator, desc.keyspace, desc.columnFamily); Tracing.traceRepair("Sending completed merkle tree to {} for {}.{}", initiator, desc.keyspace, desc.columnFamily); } MessagingService.instance().sendOneWay(new ValidationComplete(desc, trees).createMessage(), initiator); diff --git a/src/java/org/apache/cassandra/repair/consistent/SyncStatSummary.java b/src/java/org/apache/cassandra/repair/consistent/SyncStatSummary.java new file mode 100644 index 0000000000..015b55855b --- /dev/null +++ b/src/java/org/apache/cassandra/repair/consistent/SyncStatSummary.java @@ -0,0 +1,233 @@ +/* + * 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.consistent; + +import java.net.InetAddress; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +import com.google.common.collect.Lists; + +import org.apache.cassandra.repair.RepairResult; +import org.apache.cassandra.repair.RepairSessionResult; +import org.apache.cassandra.repair.SyncStat; +import org.apache.cassandra.streaming.SessionSummary; +import org.apache.cassandra.streaming.StreamSummary; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; + +import static com.google.common.collect.Iterables.filter; + +public class SyncStatSummary +{ + + private static class Session + { + final InetAddress src; + final InetAddress dst; + + int files = 0; + long bytes = 0; + long ranges = 0; + + Session(InetAddress src, InetAddress dst) + { + this.src = src; + this.dst = dst; + } + + void consumeSummary(StreamSummary summary) + { + files += summary.files; + bytes += summary.totalSize; + } + + void consumeSummaries(Collection summaries, long numRanges) + { + summaries.forEach(this::consumeSummary); + ranges += numRanges; + } + + public String toString() + { + return String.format("%s -> %s: %s ranges, %s sstables, %s bytes", src, dst, ranges, files, FBUtilities.prettyPrintMemory(bytes)); + } + } + + private static class Table + { + final String keyspace; + + final String table; + + int files = -1; + long bytes = -1; + int ranges = -1; + boolean totalsCalculated = false; + + final Map, Session> sessions = new HashMap<>(); + + Table(String keyspace, String table) + { + this.keyspace = keyspace; + this.table = table; + } + + Session getOrCreate(InetAddress from, InetAddress to) + { + Pair k = Pair.create(from, to); + if (!sessions.containsKey(k)) + { + sessions.put(k, new Session(from, to)); + } + return sessions.get(k); + } + + void consumeStat(SyncStat stat) + { + for (SessionSummary summary: stat.summaries) + { + getOrCreate(summary.coordinator, summary.peer).consumeSummaries(summary.sendingSummaries, stat.numberOfDifferences); + getOrCreate(summary.peer, summary.coordinator).consumeSummaries(summary.receivingSummaries, stat.numberOfDifferences); + } + } + + void consumeStats(List stats) + { + filter(stats, s -> s.summaries != null).forEach(this::consumeStat); + } + + void calculateTotals() + { + files = 0; + bytes = 0; + ranges = 0; + for (Session session: sessions.values()) + { + files += session.files; + bytes += session.bytes; + ranges += session.ranges; + } + totalsCalculated = true; + } + + public String toString() + { + if (!totalsCalculated) + { + calculateTotals(); + } + StringBuilder output = new StringBuilder(); + + output.append(String.format("%s.%s - %s ranges, %s sstables, %s bytes\n", keyspace, table, ranges, files, FBUtilities.prettyPrintMemory(bytes))); + for (Session session: sessions.values()) + { + output.append(" ").append(session.toString()).append('\n'); + } + return output.toString(); + } + } + + private Map, Table> summaries = new HashMap<>(); + private final boolean isEstimate; + + private int files = -1; + private long bytes = -1; + private int ranges = -1; + private boolean totalsCalculated = false; + + public SyncStatSummary(boolean isEstimate) + { + this.isEstimate = isEstimate; + } + + public void consumeRepairResult(RepairResult result) + { + Pair cf = Pair.create(result.desc.keyspace, result.desc.columnFamily); + if (!summaries.containsKey(cf)) + { + summaries.put(cf, new Table(cf.left, cf.right)); + } + summaries.get(cf).consumeStats(result.stats); + } + + public void consumeSessionResults(List results) + { + if (results != null) + { + filter(results, Objects::nonNull).forEach(r -> filter(r.repairJobResults, Objects::nonNull).forEach(this::consumeRepairResult)); + } + } + + public boolean isEmpty() + { + calculateTotals(); + return files == 0 && bytes == 0 && ranges == 0; + } + + private void calculateTotals() + { + files = 0; + bytes = 0; + ranges = 0; + summaries.values().forEach(Table::calculateTotals); + for (Table table: summaries.values()) + { + table.calculateTotals(); + files += table.files; + bytes += table.bytes; + ranges += table.ranges; + } + totalsCalculated = true; + } + + public String toString() + { + List> tables = Lists.newArrayList(summaries.keySet()); + tables.sort((o1, o2) -> + { + int ks = o1.left.compareTo(o2.left); + return ks != 0 ? ks : o1.right.compareTo(o2.right); + }); + + calculateTotals(); + + StringBuilder output = new StringBuilder(); + + if (isEstimate) + { + output.append(String.format("Total estimated streaming: %s ranges, %s sstables, %s bytes\n", ranges, files, FBUtilities.prettyPrintMemory(bytes))); + } + else + { + output.append(String.format("Total streaming: %s ranges, %s sstables, %s bytes\n", ranges, files, FBUtilities.prettyPrintMemory(bytes))); + } + + for (Pair tableName: tables) + { + Table table = summaries.get(tableName); + output.append(table.toString()).append('\n'); + } + + return output.toString(); + } +} diff --git a/src/java/org/apache/cassandra/repair/messages/PrepareMessage.java b/src/java/org/apache/cassandra/repair/messages/PrepareMessage.java index 9903114fa1..4d59942b8d 100644 --- a/src/java/org/apache/cassandra/repair/messages/PrepareMessage.java +++ b/src/java/org/apache/cassandra/repair/messages/PrepareMessage.java @@ -31,6 +31,7 @@ import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.utils.UUIDSerializer; @@ -44,8 +45,9 @@ public class PrepareMessage extends RepairMessage public final boolean isIncremental; public final long timestamp; public final boolean isGlobal; + public final PreviewKind previewKind; - public PrepareMessage(UUID parentRepairSession, List tableIds, Collection> ranges, boolean isIncremental, long timestamp, boolean isGlobal) + public PrepareMessage(UUID parentRepairSession, List tableIds, Collection> ranges, boolean isIncremental, long timestamp, boolean isGlobal, PreviewKind previewKind) { super(Type.PREPARE_MESSAGE, null); this.parentRepairSession = parentRepairSession; @@ -54,6 +56,7 @@ public class PrepareMessage extends RepairMessage this.isIncremental = isIncremental; this.timestamp = timestamp; this.isGlobal = isGlobal; + this.previewKind = previewKind; } @Override @@ -66,6 +69,7 @@ public class PrepareMessage extends RepairMessage parentRepairSession.equals(other.parentRepairSession) && isIncremental == other.isIncremental && isGlobal == other.isGlobal && + previewKind == other.previewKind && timestamp == other.timestamp && tableIds.equals(other.tableIds) && ranges.equals(other.ranges); @@ -74,7 +78,7 @@ public class PrepareMessage extends RepairMessage @Override public int hashCode() { - return Objects.hash(messageType, parentRepairSession, isGlobal, isIncremental, timestamp, tableIds, ranges); + return Objects.hash(messageType, parentRepairSession, isGlobal, previewKind, isIncremental, timestamp, tableIds, ranges); } public static class PrepareMessageSerializer implements MessageSerializer @@ -94,6 +98,7 @@ public class PrepareMessage extends RepairMessage out.writeBoolean(message.isIncremental); out.writeLong(message.timestamp); out.writeBoolean(message.isGlobal); + out.writeInt(message.previewKind.getSerializationVal()); } public PrepareMessage deserialize(DataInputPlus in, int version) throws IOException @@ -110,7 +115,8 @@ public class PrepareMessage extends RepairMessage boolean isIncremental = in.readBoolean(); long timestamp = in.readLong(); boolean isGlobal = in.readBoolean(); - return new PrepareMessage(parentRepairSession, tableIds, ranges, isIncremental, timestamp, isGlobal); + PreviewKind previewKind = PreviewKind.deserialize(in.readInt()); + return new PrepareMessage(parentRepairSession, tableIds, ranges, isIncremental, timestamp, isGlobal, previewKind); } public long serializedSize(PrepareMessage message, int version) @@ -126,6 +132,7 @@ public class PrepareMessage extends RepairMessage size += TypeSizes.sizeof(message.isIncremental); size += TypeSizes.sizeof(message.timestamp); size += TypeSizes.sizeof(message.isGlobal); + size += TypeSizes.sizeof(message.previewKind.getSerializationVal()); return size; } } diff --git a/src/java/org/apache/cassandra/repair/messages/RepairOption.java b/src/java/org/apache/cassandra/repair/messages/RepairOption.java index 3b13cd88c3..6c8ff9d990 100644 --- a/src/java/org/apache/cassandra/repair/messages/RepairOption.java +++ b/src/java/org/apache/cassandra/repair/messages/RepairOption.java @@ -28,6 +28,7 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.repair.RepairParallelism; import org.apache.cassandra.utils.FBUtilities; @@ -47,6 +48,7 @@ public class RepairOption public static final String TRACE_KEY = "trace"; public static final String SUB_RANGE_REPAIR_KEY = "sub_range_repair"; public static final String PULL_REPAIR_KEY = "pullRepair"; + public static final String PREVIEW = "previewKind"; // we don't want to push nodes too much for repair public static final int MAX_JOB_THREADS = 4; @@ -136,6 +138,7 @@ public class RepairOption RepairParallelism parallelism = RepairParallelism.fromName(options.get(PARALLELISM_KEY)); boolean primaryRange = Boolean.parseBoolean(options.get(PRIMARY_RANGE_KEY)); boolean incremental = Boolean.parseBoolean(options.get(INCREMENTAL_KEY)); + PreviewKind previewKind = PreviewKind.valueOf(options.getOrDefault(PREVIEW, PreviewKind.NONE.toString())); boolean trace = Boolean.parseBoolean(options.get(TRACE_KEY)); boolean pullRepair = Boolean.parseBoolean(options.get(PULL_REPAIR_KEY)); @@ -171,7 +174,7 @@ public class RepairOption } } - RepairOption option = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges, !ranges.isEmpty(), pullRepair); + RepairOption option = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges, !ranges.isEmpty(), pullRepair, previewKind); // data centers String dataCentersStr = options.get(DATACENTERS_KEY); @@ -252,13 +255,14 @@ public class RepairOption private final int jobThreads; private final boolean isSubrangeRepair; private final boolean pullRepair; + private final PreviewKind previewKind; 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(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection> ranges, boolean isSubrangeRepair, boolean pullRepair) + public RepairOption(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection> ranges, boolean isSubrangeRepair, boolean pullRepair, PreviewKind previewKind) { if (FBUtilities.isWindows && (DatabaseDescriptor.getDiskAccessMode() != Config.DiskAccessMode.standard || DatabaseDescriptor.getIndexAccessMode() != Config.DiskAccessMode.standard) && @@ -277,6 +281,7 @@ public class RepairOption this.ranges.addAll(ranges); this.isSubrangeRepair = isSubrangeRepair; this.pullRepair = pullRepair; + this.previewKind = previewKind; } public RepairParallelism getParallelism() @@ -339,6 +344,16 @@ public class RepairOption return isSubrangeRepair; } + public PreviewKind getPreviewKind() + { + return previewKind; + } + + public boolean isPreview() + { + return previewKind.isPreview(); + } + public boolean isInLocalDCOnly() { return dataCenters.size() == 1 && dataCenters.contains(DatabaseDescriptor.getLocalDataCenter()); } @@ -347,16 +362,17 @@ public class RepairOption public String toString() { return "repair options (" + - "parallelism: " + parallelism + - ", primary range: " + primaryRange + - ", incremental: " + incremental + - ", job threads: " + jobThreads + - ", ColumnFamilies: " + columnFamilies + - ", dataCenters: " + dataCenters + - ", hosts: " + hosts + - ", # of ranges: " + ranges.size() + - ", pull repair: " + pullRepair + - ')'; + "parallelism: " + parallelism + + ", primary range: " + primaryRange + + ", incremental: " + incremental + + ", job threads: " + jobThreads + + ", ColumnFamilies: " + columnFamilies + + ", dataCenters: " + dataCenters + + ", hosts: " + hosts + + ", previewKind: " + previewKind + + ", # of ranges: " + ranges.size() + + ", pull repair: " + pullRepair + + ')'; } public Map asMap() @@ -373,6 +389,7 @@ public class RepairOption options.put(TRACE_KEY, Boolean.toString(trace)); options.put(RANGES_KEY, Joiner.on(",").join(ranges)); options.put(PULL_REPAIR_KEY, Boolean.toString(pullRepair)); + options.put(PREVIEW, previewKind.toString()); return options; } } diff --git a/src/java/org/apache/cassandra/repair/messages/SyncComplete.java b/src/java/org/apache/cassandra/repair/messages/SyncComplete.java index 178e7102d1..7b68daf6bd 100644 --- a/src/java/org/apache/cassandra/repair/messages/SyncComplete.java +++ b/src/java/org/apache/cassandra/repair/messages/SyncComplete.java @@ -19,6 +19,8 @@ package org.apache.cassandra.repair.messages; import java.io.IOException; import java.net.InetAddress; +import java.util.ArrayList; +import java.util.List; import java.util.Objects; import org.apache.cassandra.db.TypeSizes; @@ -26,6 +28,7 @@ import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.repair.NodePair; import org.apache.cassandra.repair.RepairJobDesc; +import org.apache.cassandra.streaming.SessionSummary; /** * @@ -40,16 +43,20 @@ public class SyncComplete extends RepairMessage /** true if sync success, false otherwise */ public final boolean success; - public SyncComplete(RepairJobDesc desc, NodePair nodes, boolean success) + public final List summaries; + + public SyncComplete(RepairJobDesc desc, NodePair nodes, boolean success, List summaries) { super(Type.SYNC_COMPLETE, desc); this.nodes = nodes; this.success = success; + this.summaries = summaries; } - public SyncComplete(RepairJobDesc desc, InetAddress endpoint1, InetAddress endpoint2, boolean success) + public SyncComplete(RepairJobDesc desc, InetAddress endpoint1, InetAddress endpoint2, boolean success, List summaries) { super(Type.SYNC_COMPLETE, desc); + this.summaries = summaries; this.nodes = new NodePair(endpoint1, endpoint2); this.success = success; } @@ -63,13 +70,14 @@ public class SyncComplete extends RepairMessage return messageType == other.messageType && desc.equals(other.desc) && success == other.success && - nodes.equals(other.nodes); + nodes.equals(other.nodes) && + summaries.equals(other.summaries); } @Override public int hashCode() { - return Objects.hash(messageType, desc, success, nodes); + return Objects.hash(messageType, desc, success, nodes, summaries); } private static class SyncCompleteSerializer implements MessageSerializer @@ -79,13 +87,28 @@ public class SyncComplete extends RepairMessage RepairJobDesc.serializer.serialize(message.desc, out, version); NodePair.serializer.serialize(message.nodes, out, version); out.writeBoolean(message.success); + + out.writeInt(message.summaries.size()); + for (SessionSummary summary: message.summaries) + { + SessionSummary.serializer.serialize(summary, out, version); + } } public SyncComplete deserialize(DataInputPlus in, int version) throws IOException { RepairJobDesc desc = RepairJobDesc.serializer.deserialize(in, version); NodePair nodes = NodePair.serializer.deserialize(in, version); - return new SyncComplete(desc, nodes, in.readBoolean()); + boolean success = in.readBoolean(); + + int numSummaries = in.readInt(); + List summaries = new ArrayList<>(numSummaries); + for (int i=0; i> ranges; + public final PreviewKind previewKind; - public SyncRequest(RepairJobDesc desc, InetAddress initiator, InetAddress src, InetAddress dst, Collection> ranges) + public SyncRequest(RepairJobDesc desc, InetAddress initiator, InetAddress src, InetAddress dst, Collection> ranges, PreviewKind previewKind) { super(Type.SYNC_REQUEST, desc); this.initiator = initiator; this.src = src; this.dst = dst; this.ranges = ranges; + this.previewKind = previewKind; } @Override @@ -69,13 +72,14 @@ public class SyncRequest extends RepairMessage initiator.equals(req.initiator) && src.equals(req.src) && dst.equals(req.dst) && - ranges.equals(req.ranges); + ranges.equals(req.ranges) && + previewKind == req.previewKind; } @Override public int hashCode() { - return Objects.hash(messageType, desc, initiator, src, dst, ranges); + return Objects.hash(messageType, desc, initiator, src, dst, ranges, previewKind); } public static class SyncRequestSerializer implements MessageSerializer @@ -92,6 +96,7 @@ public class SyncRequest extends RepairMessage MessagingService.validatePartitioner(range); AbstractBounds.tokenSerializer.serialize(range, out, version); } + out.writeInt(message.previewKind.getSerializationVal()); } public SyncRequest deserialize(DataInputPlus in, int version) throws IOException @@ -104,7 +109,8 @@ public class SyncRequest extends RepairMessage List> ranges = new ArrayList<>(rangesCount); for (int i = 0; i < rangesCount; ++i) ranges.add((Range) AbstractBounds.tokenSerializer.deserialize(in, MessagingService.globalPartitioner(), version)); - return new SyncRequest(desc, owner, src, dst, ranges); + PreviewKind previewKind = PreviewKind.deserialize(in.readInt()); + return new SyncRequest(desc, owner, src, dst, ranges, previewKind); } public long serializedSize(SyncRequest message, int version) @@ -114,6 +120,7 @@ public class SyncRequest extends RepairMessage size += TypeSizes.sizeof(message.ranges.size()); for (Range range : message.ranges) size += AbstractBounds.tokenSerializer.serializedSize(range, version); + size += TypeSizes.sizeof(message.previewKind.getSerializationVal()); return size; } } @@ -126,6 +133,7 @@ public class SyncRequest extends RepairMessage ", src=" + src + ", dst=" + dst + ", ranges=" + ranges + + ", previewKind=" + previewKind + "} " + super.toString(); } } diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java index fd98b37695..aadf7c1457 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@ -61,6 +61,7 @@ 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.streaming.PreviewKind; import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.RepairParallelism; import org.apache.cassandra.repair.RepairSession; @@ -167,6 +168,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai Set endpoints, boolean isConsistent, boolean pullRepair, + PreviewKind previewKind, ListeningExecutorService executor, String... cfnames) { @@ -176,7 +178,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai if (cfnames.length == 0) return null; - final RepairSession session = new RepairSession(parentRepairSession, UUIDGen.getTimeUUID(), range, keyspace, parallelismDegree, endpoints, isConsistent, pullRepair, cfnames); + final RepairSession session = new RepairSession(parentRepairSession, UUIDGen.getTimeUUID(), range, keyspace, parallelismDegree, endpoints, isConsistent, pullRepair, previewKind, cfnames); sessions.put(session.getId(), session); // register listeners @@ -319,7 +321,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai { // we only want repairedAt for incremental repairs, for non incremental repairs, UNREPAIRED_SSTABLE will preserve repairedAt on streamed sstables long repairedAt = options.isIncremental() ? Clock.instance.currentTimeMillis() : ActiveRepairService.UNREPAIRED_SSTABLE; - registerParentRepairSession(parentRepairSession, coordinator, columnFamilyStores, options.getRanges(), options.isIncremental(), repairedAt, options.isGlobal()); + registerParentRepairSession(parentRepairSession, coordinator, columnFamilyStores, options.getRanges(), options.isIncremental(), repairedAt, options.isGlobal(), options.getPreviewKind()); final CountDownLatch prepareLatch = new CountDownLatch(endpoints.size()); final AtomicBoolean status = new AtomicBoolean(true); final Set failedNodes = Collections.synchronizedSet(new HashSet()); @@ -351,7 +353,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai { if (FailureDetector.instance.isAlive(neighbour)) { - PrepareMessage message = new PrepareMessage(parentRepairSession, tableIds, options.getRanges(), options.isIncremental(), repairedAt, options.isGlobal()); + PrepareMessage message = new PrepareMessage(parentRepairSession, tableIds, options.getRanges(), options.isIncremental(), repairedAt, options.isGlobal(), options.getPreviewKind()); MessageOut msg = message.createMessage(); MessagingService.instance().sendRR(msg, neighbour, callback, DatabaseDescriptor.getRpcTimeout(), true); } @@ -386,7 +388,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai throw new RuntimeException(errorMsg); } - public void registerParentRepairSession(UUID parentRepairSession, InetAddress coordinator, List columnFamilyStores, Collection> ranges, boolean isIncremental, long repairedAt, boolean isGlobal) + public void registerParentRepairSession(UUID parentRepairSession, InetAddress coordinator, List columnFamilyStores, Collection> ranges, boolean isIncremental, long repairedAt, boolean isGlobal, PreviewKind previewKind) { assert isIncremental || repairedAt == ActiveRepairService.UNREPAIRED_SSTABLE; if (!registeredForEndpointChanges) @@ -396,7 +398,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai registeredForEndpointChanges = true; } - parentRepairSessions.put(parentRepairSession, new ParentRepairSession(coordinator, columnFamilyStores, ranges, isIncremental, repairedAt, isGlobal)); + parentRepairSessions.put(parentRepairSession, new ParentRepairSession(coordinator, columnFamilyStores, ranges, isIncremental, repairedAt, isGlobal, previewKind)); } public ParentRepairSession getParentRepairSession(UUID parentSessionId) @@ -444,7 +446,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai case SYNC_COMPLETE: // one of replica is synced. SyncComplete sync = (SyncComplete) message; - session.syncComplete(desc, sync.nodes, sync.success); + session.syncComplete(desc, sync.nodes, sync.success, sync.summaries); break; default: break; @@ -464,8 +466,9 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai public final boolean isGlobal; public final long repairedAt; public final InetAddress coordinator; + public final PreviewKind previewKind; - public ParentRepairSession(InetAddress coordinator, List columnFamilyStores, Collection> ranges, boolean isIncremental, long repairedAt, boolean isGlobal) + public ParentRepairSession(InetAddress coordinator, List columnFamilyStores, Collection> ranges, boolean isIncremental, long repairedAt, boolean isGlobal, PreviewKind previewKind) { this.coordinator = coordinator; for (ColumnFamilyStore cfs : columnFamilyStores) @@ -476,6 +479,27 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai this.repairedAt = repairedAt; this.isIncremental = isIncremental; this.isGlobal = isGlobal; + this.previewKind = previewKind; + } + + public boolean isPreview() + { + return previewKind != PreviewKind.NONE; + } + + public Predicate getPreviewPredicate() + { + switch (previewKind) + { + case ALL: + return (s) -> true; + case REPAIRED: + return (s) -> s.isRepaired(); + case UNREPAIRED: + return (s) -> !s.isRepaired(); + default: + throw new RuntimeException("Can't get preview predicate for preview kind " + previewKind); + } } public synchronized void maybeSnapshot(TableId tableId, UUID parentSessionId) diff --git a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java index 86340a5355..5f734c9886 100644 --- a/src/java/org/apache/cassandra/streaming/ConnectionHandler.java +++ b/src/java/org/apache/cassandra/streaming/ConnectionHandler.java @@ -64,10 +64,12 @@ public class ConnectionHandler private IncomingMessageHandler incoming; private OutgoingMessageHandler outgoing; + private final boolean isPreview; - ConnectionHandler(StreamSession session, int incomingSocketTimeout) + ConnectionHandler(StreamSession session, int incomingSocketTimeout, boolean isPreview) { this.session = session; + this.isPreview = isPreview; this.incoming = new IncomingMessageHandler(session, incomingSocketTimeout); this.outgoing = new OutgoingMessageHandler(session); } @@ -142,6 +144,9 @@ public class ConnectionHandler if (outgoing.isClosed()) throw new RuntimeException("Outgoing stream handler has been closed"); + if (message.type == StreamMessage.Type.FILE && isPreview) + throw new RuntimeException("Cannot send file messages for preview streaming sessions"); + outgoing.enqueue(message); } @@ -191,14 +196,14 @@ public class ConnectionHandler @SuppressWarnings("resource") private void sendInitMessage() throws IOException { - StreamInitMessage message = new StreamInitMessage( - FBUtilities.getBroadcastAddress(), - session.sessionIndex(), - session.planId(), - session.streamOperation(), - !isOutgoingHandler, - session.keepSSTableLevel(), - session.getPendingRepair()); + StreamInitMessage message = new StreamInitMessage(FBUtilities.getBroadcastAddress(), + session.sessionIndex(), + session.planId(), + session.streamOperation(), + !isOutgoingHandler, + session.keepSSTableLevel(), + session.getPendingRepair(), + session.getPreviewKind()); ByteBuffer messageBuf = message.createMessage(false, protocolVersion); DataOutputStreamPlus out = getWriteChannel(socket); out.write(messageBuf); diff --git a/src/java/org/apache/cassandra/streaming/PreviewKind.java b/src/java/org/apache/cassandra/streaming/PreviewKind.java new file mode 100644 index 0000000000..3b4d2a08ad --- /dev/null +++ b/src/java/org/apache/cassandra/streaming/PreviewKind.java @@ -0,0 +1,76 @@ +/* + * 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.streaming; + + +import java.util.UUID; + +import com.google.common.base.Predicate; +import com.google.common.base.Predicates; + +import org.apache.cassandra.io.sstable.format.SSTableReader; + +public enum PreviewKind +{ + NONE(0, null), + ALL(1, Predicates.alwaysTrue()), + UNREPAIRED(2, Predicates.not(SSTableReader::isRepaired)), + REPAIRED(3, SSTableReader::isRepaired); + + private final int serializationVal; + private final Predicate streamingPredicate; + + PreviewKind(int serializationVal, Predicate streamingPredicate) + { + assert ordinal() == serializationVal; + this.serializationVal = serializationVal; + this.streamingPredicate = streamingPredicate; + } + + public int getSerializationVal() + { + return serializationVal; + } + + public static PreviewKind deserialize(int serializationVal) + { + return values()[serializationVal]; + } + + public Predicate getStreamingPredicate() + { + return streamingPredicate; + } + + public boolean isPreview() + { + return this != NONE; + } + + public String logPrefix() + { + return isPreview() ? "preview repair" : "repair"; + } + + public String logPrefix(UUID sessionId) + { + return '[' + logPrefix() + " #" + sessionId.toString() + ']'; + } + +} diff --git a/src/java/org/apache/cassandra/streaming/SessionInfo.java b/src/java/org/apache/cassandra/streaming/SessionInfo.java index 3bcb20cbbd..1521614e0b 100644 --- a/src/java/org/apache/cassandra/streaming/SessionInfo.java +++ b/src/java/org/apache/cassandra/streaming/SessionInfo.java @@ -27,6 +27,8 @@ import com.google.common.base.Predicate; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; +import org.apache.cassandra.utils.FBUtilities; + /** * Stream session info. */ @@ -190,4 +192,9 @@ public final class SessionInfo implements Serializable }); return Iterables.size(completed); } + + public SessionSummary createSummary() + { + return new SessionSummary(FBUtilities.getBroadcastAddress(), peer, receivingSummaries, sendingSummaries); + } } diff --git a/src/java/org/apache/cassandra/streaming/SessionSummary.java b/src/java/org/apache/cassandra/streaming/SessionSummary.java new file mode 100644 index 0000000000..d52c2ca360 --- /dev/null +++ b/src/java/org/apache/cassandra/streaming/SessionSummary.java @@ -0,0 +1,141 @@ +/* + * 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.streaming; + +import java.io.IOException; +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.io.IVersionedSerializer; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.serializers.InetAddressSerializer; +import org.apache.cassandra.utils.ByteBufferUtil; + +public class SessionSummary +{ + public final InetAddress coordinator; + public final InetAddress peer; + /** Immutable collection of receiving summaries */ + public final Collection receivingSummaries; + /** Immutable collection of sending summaries*/ + public final Collection sendingSummaries; + + public SessionSummary(InetAddress coordinator, InetAddress peer, + Collection receivingSummaries, + Collection sendingSummaries) + { + assert coordinator != null; + assert peer != null; + assert receivingSummaries != null; + assert sendingSummaries != null; + + this.coordinator = coordinator; + this.peer = peer; + this.receivingSummaries = receivingSummaries; + this.sendingSummaries = sendingSummaries; + } + + public boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + SessionSummary summary = (SessionSummary) o; + + if (!coordinator.equals(summary.coordinator)) return false; + if (!peer.equals(summary.peer)) return false; + if (!receivingSummaries.equals(summary.receivingSummaries)) return false; + return sendingSummaries.equals(summary.sendingSummaries); + } + + public int hashCode() + { + int result = coordinator.hashCode(); + result = 31 * result + peer.hashCode(); + result = 31 * result + receivingSummaries.hashCode(); + result = 31 * result + sendingSummaries.hashCode(); + return result; + } + + public static IVersionedSerializer serializer = new IVersionedSerializer() + { + public void serialize(SessionSummary summary, DataOutputPlus out, int version) throws IOException + { + ByteBufferUtil.writeWithLength(InetAddressSerializer.instance.serialize(summary.coordinator), out); + ByteBufferUtil.writeWithLength(InetAddressSerializer.instance.serialize(summary.peer), out); + + out.writeInt(summary.receivingSummaries.size()); + for (StreamSummary streamSummary: summary.receivingSummaries) + { + StreamSummary.serializer.serialize(streamSummary, out, version); + } + + out.writeInt(summary.sendingSummaries.size()); + for (StreamSummary streamSummary: summary.sendingSummaries) + { + StreamSummary.serializer.serialize(streamSummary, out, version); + } + } + + public SessionSummary deserialize(DataInputPlus in, int version) throws IOException + { + InetAddress coordinator = InetAddressSerializer.instance.deserialize(ByteBufferUtil.readWithLength(in)); + InetAddress peer = InetAddressSerializer.instance.deserialize(ByteBufferUtil.readWithLength(in)); + + int numRcvd = in.readInt(); + List receivingSummaries = new ArrayList<>(numRcvd); + for (int i=0; i sendingSummaries = new ArrayList<>(numRcvd); + for (int i=0; i sessionsToConnect = null; private final UUID pendingRepair; + private final PreviewKind previewKind; public StreamCoordinator(int connectionsPerHost, boolean keepSSTableLevel, StreamConnectionFactory factory, - boolean connectSequentially, UUID pendingRepair) + boolean connectSequentially, UUID pendingRepair, PreviewKind previewKind) { this.connectionsPerHost = connectionsPerHost; this.factory = factory; this.keepSSTableLevel = keepSSTableLevel; this.connectSequentially = connectSequentially; this.pendingRepair = pendingRepair; + this.previewKind = previewKind; } public void setConnectionFactory(StreamConnectionFactory factory) @@ -293,7 +295,7 @@ public class StreamCoordinator // create if (streamSessions.size() < connectionsPerHost) { - StreamSession session = new StreamSession(peer, connecting, factory, streamSessions.size(), keepSSTableLevel, pendingRepair); + StreamSession session = new StreamSession(peer, connecting, factory, streamSessions.size(), keepSSTableLevel, pendingRepair, previewKind); streamSessions.put(++lastReturned, session); return session; } @@ -325,7 +327,7 @@ public class StreamCoordinator StreamSession session = streamSessions.get(id); if (session == null) { - session = new StreamSession(peer, connecting, factory, id, keepSSTableLevel, pendingRepair); + session = new StreamSession(peer, connecting, factory, id, keepSSTableLevel, pendingRepair, previewKind); streamSessions.put(id, session); } return session; diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java index b5a6214d13..05a8d30974 100644 --- a/src/java/org/apache/cassandra/streaming/StreamPlan.java +++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java @@ -22,9 +22,10 @@ import java.util.*; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; -import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.utils.UUIDGen; +import static org.apache.cassandra.service.ActiveRepairService.NO_PENDING_REPAIR; + /** * {@link StreamPlan} is a helper class that builds StreamOperation of given configuration. * @@ -47,20 +48,20 @@ public class StreamPlan */ public StreamPlan(StreamOperation streamOperation) { - this(streamOperation, 1, false, false, null); + this(streamOperation, 1, false, false, NO_PENDING_REPAIR, PreviewKind.NONE); } public StreamPlan(StreamOperation streamOperation, boolean keepSSTableLevels, boolean connectSequentially) { - this(streamOperation, 1, keepSSTableLevels, connectSequentially, null); + this(streamOperation, 1, keepSSTableLevels, connectSequentially, NO_PENDING_REPAIR, PreviewKind.NONE); } public StreamPlan(StreamOperation streamOperation, int connectionsPerHost, boolean keepSSTableLevels, - boolean connectSequentially, UUID pendingRepair) + boolean connectSequentially, UUID pendingRepair, PreviewKind previewKind) { this.streamOperation = streamOperation; this.coordinator = new StreamCoordinator(connectionsPerHost, keepSSTableLevels, new DefaultConnectionFactory(), - connectSequentially, pendingRepair); + connectSequentially, pendingRepair, previewKind); } /** diff --git a/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java b/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java index b7e475ac3c..34e7cc8b63 100644 --- a/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java +++ b/src/java/org/apache/cassandra/streaming/StreamReceiveTask.java @@ -24,6 +24,7 @@ import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import com.google.common.base.Preconditions; import com.google.common.collect.Iterables; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -92,6 +93,8 @@ public class StreamReceiveTask extends StreamTask */ public synchronized void received(SSTableMultiWriter sstable) { + Preconditions.checkState(!session.isPreview(), "we should never receive sstables when previewing"); + if (done) { logger.warn("[{}] Received sstable {} on already finished stream received task. Aborting sstable.", session.planId(), diff --git a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java index 7845986398..67d7d0dff8 100644 --- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java +++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java @@ -71,9 +71,9 @@ public final class StreamResultFuture extends AbstractFuture set(getCurrentState()); } - private StreamResultFuture(UUID planId, StreamOperation streamOperation, boolean keepSSTableLevels, UUID pendingRepair) + private StreamResultFuture(UUID planId, StreamOperation streamOperation, boolean keepSSTableLevels, UUID pendingRepair, PreviewKind previewKind) { - this(planId, streamOperation, new StreamCoordinator(0, keepSSTableLevels, new DefaultConnectionFactory(), false, pendingRepair)); + this(planId, streamOperation, new StreamCoordinator(0, keepSSTableLevels, new DefaultConnectionFactory(), false, pendingRepair, previewKind)); } static StreamResultFuture init(UUID planId, StreamOperation streamOperation, Collection listeners, @@ -107,7 +107,8 @@ public final class StreamResultFuture extends AbstractFuture boolean isForOutgoing, int version, boolean keepSSTableLevel, - UUID pendingRepair) throws IOException + UUID pendingRepair, + PreviewKind previewKind) throws IOException { StreamResultFuture future = StreamManager.instance.getReceivingStream(planId); if (future == null) @@ -115,7 +116,7 @@ public final class StreamResultFuture extends AbstractFuture logger.info("[Stream #{} ID#{}] Creating new streaming plan for {}", planId, sessionIndex, streamOperation.getDescription()); // The main reason we create a StreamResultFuture on the receiving side is for JMX exposure. - future = new StreamResultFuture(planId, streamOperation, keepSSTableLevel, pendingRepair); + future = new StreamResultFuture(planId, streamOperation, keepSSTableLevel, pendingRepair, previewKind); StreamManager.instance.registerReceiving(future); } future.attachConnection(from, sessionIndex, connection, isForOutgoing, version); diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java index adb8e791ab..5ca99387e4 100644 --- a/src/java/org/apache/cassandra/streaming/StreamSession.java +++ b/src/java/org/apache/cassandra/streaming/StreamSession.java @@ -165,6 +165,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber private final boolean keepSSTableLevel; private ScheduledFuture keepAliveFuture = null; private final UUID pendingRepair; + private final PreviewKind previewKind; public static enum State { @@ -181,12 +182,11 @@ public class StreamSession implements IEndpointStateChangeSubscriber /** * Create new streaming session with the peer. - * - * @param peer Address of streaming peer + * @param peer Address of streaming peer * @param connecting Actual connecting address * @param factory is used for establishing connection */ - public StreamSession(InetAddress peer, InetAddress connecting, StreamConnectionFactory factory, int index, boolean keepSSTableLevel, UUID pendingRepair) + public StreamSession(InetAddress peer, InetAddress connecting, StreamConnectionFactory factory, int index, boolean keepSSTableLevel, UUID pendingRepair, PreviewKind previewKind) { this.peer = peer; this.connecting = connecting; @@ -194,10 +194,11 @@ public class StreamSession implements IEndpointStateChangeSubscriber this.factory = factory; this.handler = new ConnectionHandler(this, isKeepAliveSupported()? (int)TimeUnit.SECONDS.toMillis(2 * DatabaseDescriptor.getStreamingKeepAlivePeriod()) : - DatabaseDescriptor.getStreamingSocketTimeout()); + DatabaseDescriptor.getStreamingSocketTimeout(), previewKind.isPreview()); this.metrics = StreamingMetrics.get(connecting); this.keepSSTableLevel = keepSSTableLevel; this.pendingRepair = pendingRepair; + this.previewKind = previewKind; } public UUID planId() @@ -225,6 +226,16 @@ public class StreamSession implements IEndpointStateChangeSubscriber return pendingRepair; } + public boolean isPreview() + { + return previewKind.isPreview(); + } + + public PreviewKind getPreviewKind() + { + return previewKind; + } + public LifecycleTransaction getTransaction(TableId tableId) { assert receivers.containsKey(tableId); @@ -314,7 +325,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber flushSSTables(stores); List> normalizedRanges = Range.normalize(ranges); - List sections = getSSTableSectionsForRanges(normalizedRanges, stores, pendingRepair); + List sections = getSSTableSectionsForRanges(normalizedRanges, stores, pendingRepair, previewKind); try { addTransferFiles(sections); @@ -356,7 +367,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber } @VisibleForTesting - public static List getSSTableSectionsForRanges(Collection> ranges, Collection stores, UUID pendingRepair) + public static List getSSTableSectionsForRanges(Collection> ranges, Collection stores, UUID pendingRepair, PreviewKind previewKind) { Refs refs = new Refs<>(); try @@ -370,7 +381,11 @@ public class StreamSession implements IEndpointStateChangeSubscriber Set sstables = Sets.newHashSet(); SSTableIntervalTree intervalTree = SSTableIntervalTree.build(view.select(SSTableSet.CANONICAL)); Predicate predicate; - if (pendingRepair == ActiveRepairService.NO_PENDING_REPAIR) + if (previewKind.isPreview()) + { + predicate = previewKind.getStreamingPredicate(); + } + else if (pendingRepair == ActiveRepairService.NO_PENDING_REPAIR) { predicate = Predicates.alwaysTrue(); } @@ -620,6 +635,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber handler.sendMessage(prepare); } + if (isPreview()) + { + completePreview(); + return; + } + // if there are files to stream if (!maybeCompleted()) startStreamingFiles(); @@ -650,6 +671,11 @@ public class StreamSession implements IEndpointStateChangeSubscriber */ public void receive(IncomingFileMessage message) { + if (isPreview()) + { + throw new RuntimeException("Cannot receive files for preview session"); + } + long headerSize = message.header.size(); StreamingMetrics.totalIncomingBytes.inc(headerSize); metrics.incomingBytes.inc(headerSize); @@ -753,6 +779,22 @@ public class StreamSession implements IEndpointStateChangeSubscriber closeSession(State.FAILED); } + private void completePreview() + { + try + { + state(State.WAIT_COMPLETE); + closeSession(State.COMPLETE); + } + finally + { + // aborting the tasks here needs to be the last thing we do so that we + // accurately report expected streaming, but don't leak any sstable refs + for (StreamTask task : Iterables.concat(receivers.values(), transfers.values())) + task.abort(); + } + } + private boolean maybeCompleted() { boolean completed = receivers.isEmpty() && transfers.isEmpty(); @@ -803,6 +845,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber streamResult.handleSessionPrepared(this); state(State.STREAMING); + for (StreamTransferTask task : transfers.values()) { Collection messages = task.getFileMessages(); diff --git a/src/java/org/apache/cassandra/streaming/StreamState.java b/src/java/org/apache/cassandra/streaming/StreamState.java index 4ee3c8d698..be376775ea 100644 --- a/src/java/org/apache/cassandra/streaming/StreamState.java +++ b/src/java/org/apache/cassandra/streaming/StreamState.java @@ -18,11 +18,13 @@ package org.apache.cassandra.streaming; import java.io.Serializable; +import java.util.List; import java.util.Set; import java.util.UUID; import com.google.common.base.Predicate; import com.google.common.collect.Iterables; +import com.google.common.collect.Lists; /** * Current snapshot of streaming progress. @@ -50,4 +52,9 @@ public class StreamState implements Serializable } }); } + + public List createSummaries() + { + return Lists.newArrayList(Iterables.transform(sessions, SessionInfo::createSummary)); + } } diff --git a/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java b/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java index 4619561ab0..ceaa4d1448 100644 --- a/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java +++ b/src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java @@ -31,6 +31,7 @@ import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.net.CompactEndpointSerializationHelper; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.streaming.StreamOperation; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.utils.UUIDSerializer; /** @@ -50,8 +51,9 @@ public class StreamInitMessage public final boolean isForOutgoing; public final boolean keepSSTableLevel; public final UUID pendingRepair; + public final PreviewKind previewKind; - public StreamInitMessage(InetAddress from, int sessionIndex, UUID planId, StreamOperation streamOperation, boolean isForOutgoing, boolean keepSSTableLevel, UUID pendingRepair) + public StreamInitMessage(InetAddress from, int sessionIndex, UUID planId, StreamOperation streamOperation, boolean isForOutgoing, boolean keepSSTableLevel, UUID pendingRepair, PreviewKind previewKind) { this.from = from; this.sessionIndex = sessionIndex; @@ -60,6 +62,7 @@ public class StreamInitMessage this.isForOutgoing = isForOutgoing; this.keepSSTableLevel = keepSSTableLevel; this.pendingRepair = pendingRepair; + this.previewKind = previewKind; } /** @@ -120,6 +123,7 @@ public class StreamInitMessage { UUIDSerializer.serializer.serialize(message.pendingRepair, out, MessagingService.current_version); } + out.writeInt(message.previewKind.getSerializationVal()); } public StreamInitMessage deserialize(DataInputPlus in, int version) throws IOException @@ -132,7 +136,8 @@ public class StreamInitMessage boolean keepSSTableLevel = in.readBoolean(); UUID pendingRepair = in.readBoolean() ? UUIDSerializer.serializer.deserialize(in, version) : null; - return new StreamInitMessage(from, sessionIndex, planId, StreamOperation.fromString(description), sentByInitiator, keepSSTableLevel, pendingRepair); + PreviewKind previewKind = PreviewKind.deserialize(in.readInt()); + return new StreamInitMessage(from, sessionIndex, planId, StreamOperation.fromString(description), sentByInitiator, keepSSTableLevel, pendingRepair, previewKind); } public long serializedSize(StreamInitMessage message, int version) @@ -148,6 +153,7 @@ public class StreamInitMessage { size += UUIDSerializer.serializer.serializedSize(message.pendingRepair, MessagingService.current_version); } + size += TypeSizes.sizeof(message.previewKind.getSerializationVal()); return size; } } diff --git a/src/java/org/apache/cassandra/tools/nodetool/Repair.java b/src/java/org/apache/cassandra/tools/nodetool/Repair.java index 48f929f7c4..317a677058 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/Repair.java +++ b/src/java/org/apache/cassandra/tools/nodetool/Repair.java @@ -32,6 +32,7 @@ import java.util.Set; import com.google.common.collect.Sets; import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.repair.RepairParallelism; import org.apache.cassandra.repair.messages.RepairOption; import org.apache.cassandra.tools.NodeProbe; @@ -73,6 +74,12 @@ public class Repair extends NodeToolCmd @Option(title = "full", name = {"-full", "--full"}, description = "Use -full to issue a full repair.") private boolean fullRepair = false; + @Option(title = "preview", name = {"-prv", "--preview"}, description = "Determine ranges and amount of data to be streamed, but don't actually perform repair") + private boolean preview = false; + + @Option(title = "validate", name = {"-vd", "--validate"}, description = "Checks that repaired data is in sync between nodes. Out of sync repaired data indicates a full repair should be run.") + private boolean validate = 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)") @@ -84,6 +91,26 @@ public class Repair extends NodeToolCmd @Option(title = "pull_repair", name = {"-pl", "--pull"}, description = "Use --pull to perform a one way repair where data is only streamed from a remote node to this node.") private boolean pullRepair = false; + private PreviewKind getPreviewKind() + { + if (validate) + { + return PreviewKind.REPAIRED; + } + else if (preview && fullRepair) + { + return PreviewKind.ALL; + } + else if (preview) + { + return PreviewKind.UNREPAIRED; + } + else + { + return PreviewKind.NONE; + } + } + @Override public void execute(NodeProbe probe) { @@ -112,6 +139,8 @@ public class Repair extends NodeToolCmd options.put(RepairOption.TRACE_KEY, Boolean.toString(trace)); options.put(RepairOption.COLUMNFAMILIES_KEY, StringUtils.join(cfnames, ",")); options.put(RepairOption.PULL_REPAIR_KEY, Boolean.toString(pullRepair)); + options.put(RepairOption.PREVIEW, getPreviewKind().toString()); + if (!startToken.isEmpty() || !endToken.isEmpty()) { options.put(RepairOption.RANGES_KEY, startToken + ":" + endToken); diff --git a/test/data/serialization/4.0/gms.EndpointState.bin b/test/data/serialization/4.0/gms.EndpointState.bin new file mode 100644 index 0000000000..fb7d168076 Binary files /dev/null and b/test/data/serialization/4.0/gms.EndpointState.bin differ diff --git a/test/data/serialization/4.0/gms.Gossip.bin b/test/data/serialization/4.0/gms.Gossip.bin new file mode 100644 index 0000000000..af5ac57146 Binary files /dev/null and b/test/data/serialization/4.0/gms.Gossip.bin differ diff --git a/test/data/serialization/4.0/service.SyncComplete.bin b/test/data/serialization/4.0/service.SyncComplete.bin new file mode 100644 index 0000000000..ba84349bd9 Binary files /dev/null and b/test/data/serialization/4.0/service.SyncComplete.bin differ diff --git a/test/data/serialization/4.0/service.SyncRequest.bin b/test/data/serialization/4.0/service.SyncRequest.bin new file mode 100644 index 0000000000..6d688a4299 Binary files /dev/null and b/test/data/serialization/4.0/service.SyncRequest.bin differ diff --git a/test/data/serialization/4.0/service.ValidationComplete.bin b/test/data/serialization/4.0/service.ValidationComplete.bin new file mode 100644 index 0000000000..7433d64d56 Binary files /dev/null and b/test/data/serialization/4.0/service.ValidationComplete.bin differ diff --git a/test/data/serialization/4.0/service.ValidationRequest.bin b/test/data/serialization/4.0/service.ValidationRequest.bin new file mode 100644 index 0000000000..a00763ba93 Binary files /dev/null and b/test/data/serialization/4.0/service.ValidationRequest.bin differ diff --git a/test/data/serialization/4.0/utils.EstimatedHistogram.bin b/test/data/serialization/4.0/utils.EstimatedHistogram.bin new file mode 100644 index 0000000000..e878eda0ce Binary files /dev/null and b/test/data/serialization/4.0/utils.EstimatedHistogram.bin differ diff --git a/test/unit/org/apache/cassandra/AbstractSerializationsTester.java b/test/unit/org/apache/cassandra/AbstractSerializationsTester.java index 04cb08342a..3611f0e639 100644 --- a/test/unit/org/apache/cassandra/AbstractSerializationsTester.java +++ b/test/unit/org/apache/cassandra/AbstractSerializationsTester.java @@ -36,10 +36,11 @@ import java.util.Map; public class AbstractSerializationsTester { - protected static final String CUR_VER = System.getProperty("cassandra.version", "3.0"); + protected static final String CUR_VER = System.getProperty("cassandra.version", "4.0"); protected static final Map VERSION_MAP = new HashMap () {{ put("3.0", MessagingService.VERSION_30); + put("4.0", MessagingService.VERSION_40); }}; protected static final boolean EXECUTE_WRITES = Boolean.getBoolean("cassandra.test-serialization-writes"); diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionManagerGetSSTablesForValidationTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionManagerGetSSTablesForValidationTest.java index 0ee85c6e95..b9e3c17434 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionManagerGetSSTablesForValidationTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionManagerGetSSTablesForValidationTest.java @@ -41,6 +41,7 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.Validator; import org.apache.cassandra.schema.KeyspaceParams; @@ -106,7 +107,8 @@ public class CompactionManagerGetSSTablesForValidationTest Sets.newHashSet(range), incremental, incremental ? System.currentTimeMillis() : ActiveRepairService.UNREPAIRED_SSTABLE, - true); + true, + PreviewKind.NONE); desc = new RepairJobDesc(sessionID, UUIDGen.getTimeUUID(), ks, tbl, Collections.singleton(range)); } @@ -135,7 +137,7 @@ public class CompactionManagerGetSSTablesForValidationTest modifySSTables(); // get sstables for repair - Validator validator = new Validator(desc, coordinator, FBUtilities.nowInSeconds(), true); + Validator validator = new Validator(desc, coordinator, FBUtilities.nowInSeconds(), true, PreviewKind.NONE); Set sstables = Sets.newHashSet(CompactionManager.instance.getSSTablesToValidate(cfs, validator)); Assert.assertNotNull(sstables); Assert.assertEquals(1, sstables.size()); @@ -150,7 +152,7 @@ public class CompactionManagerGetSSTablesForValidationTest modifySSTables(); // get sstables for repair - Validator validator = new Validator(desc, coordinator, FBUtilities.nowInSeconds(), false); + Validator validator = new Validator(desc, coordinator, FBUtilities.nowInSeconds(), false, PreviewKind.NONE); Set sstables = Sets.newHashSet(CompactionManager.instance.getSSTablesToValidate(cfs, validator)); Assert.assertNotNull(sstables); Assert.assertEquals(2, sstables.size()); @@ -166,7 +168,7 @@ public class CompactionManagerGetSSTablesForValidationTest modifySSTables(); // get sstables for repair - Validator validator = new Validator(desc, coordinator, FBUtilities.nowInSeconds(), false); + Validator validator = new Validator(desc, coordinator, FBUtilities.nowInSeconds(), false, PreviewKind.NONE); Set sstables = Sets.newHashSet(CompactionManager.instance.getSSTablesToValidate(cfs, validator)); Assert.assertNotNull(sstables); Assert.assertEquals(3, sstables.size()); diff --git a/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java b/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java index 0a38cd974e..360a2cd699 100644 --- a/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java @@ -51,6 +51,7 @@ import org.apache.cassandra.io.sstable.ISSTableScanner; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.notifications.SSTableAddedNotification; import org.apache.cassandra.notifications.SSTableRepairStatusChanged; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.Validator; import org.apache.cassandra.schema.CompactionParams; @@ -193,9 +194,16 @@ public class LeveledCompactionStrategyTest Range range = new Range<>(Util.token(""), Util.token("")); int gcBefore = keyspace.getColumnFamilyStore(CF_STANDARDDLEVELED).gcBefore(FBUtilities.nowInSeconds()); UUID parentRepSession = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(parentRepSession, FBUtilities.getBroadcastAddress(), Arrays.asList(cfs), Arrays.asList(range), false, ActiveRepairService.UNREPAIRED_SSTABLE, true); + ActiveRepairService.instance.registerParentRepairSession(parentRepSession, + FBUtilities.getBroadcastAddress(), + Arrays.asList(cfs), + Arrays.asList(range), + false, + ActiveRepairService.UNREPAIRED_SSTABLE, + true, + PreviewKind.NONE); RepairJobDesc desc = new RepairJobDesc(parentRepSession, UUID.randomUUID(), KEYSPACE1, CF_STANDARDDLEVELED, Arrays.asList(range)); - Validator validator = new Validator(desc, FBUtilities.getBroadcastAddress(), gcBefore); + Validator validator = new Validator(desc, FBUtilities.getBroadcastAddress(), gcBefore, PreviewKind.NONE); CompactionManager.instance.submitValidation(cfs, validator).get(); } diff --git a/test/unit/org/apache/cassandra/dht/StreamStateStoreTest.java b/test/unit/org/apache/cassandra/dht/StreamStateStoreTest.java index f11362f3e5..b5f80360ec 100644 --- a/test/unit/org/apache/cassandra/dht/StreamStateStoreTest.java +++ b/test/unit/org/apache/cassandra/dht/StreamStateStoreTest.java @@ -25,6 +25,7 @@ import org.junit.Test; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.streaming.DefaultConnectionFactory; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.streaming.StreamEvent; import org.apache.cassandra.streaming.StreamSession; import org.apache.cassandra.utils.FBUtilities; @@ -50,7 +51,7 @@ public class StreamStateStoreTest Range range = new Range<>(factory.fromString("0"), factory.fromString("100")); InetAddress local = FBUtilities.getBroadcastAddress(); - StreamSession session = new StreamSession(local, local, new DefaultConnectionFactory(), 0, true, null); + StreamSession session = new StreamSession(local, local, new DefaultConnectionFactory(), 0, true, null, PreviewKind.NONE); session.addStreamRequest("keyspace1", Collections.singleton(range), Collections.singleton("cf")); StreamStateStore store = new StreamStateStore(); @@ -71,7 +72,7 @@ public class StreamStateStoreTest // add different range within the same keyspace Range range2 = new Range<>(factory.fromString("100"), factory.fromString("200")); - session = new StreamSession(local, local, new DefaultConnectionFactory(), 0, true, null); + session = new StreamSession(local, local, new DefaultConnectionFactory(), 0, true, null, PreviewKind.NONE); session.addStreamRequest("keyspace1", Collections.singleton(range2), Collections.singleton("cf")); session.state(StreamSession.State.COMPLETE); store.handleStreamEvent(new StreamEvent.SessionCompleteEvent(session)); diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java index 97bd321cbc..53f5ab3e59 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java @@ -56,6 +56,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.db.lifecycle.LifecycleTransaction; import org.apache.cassandra.metrics.StorageMetrics; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.streaming.StreamSession; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; @@ -789,7 +790,7 @@ public class SSTableRewriterTest extends SSTableWriterTestBase List sectionsBeforeRewrite = StreamSession.getSSTableSectionsForRanges( Collections.singleton(new Range(firstToken, firstToken)), - Collections.singleton(cfs), null); + Collections.singleton(cfs), null, PreviewKind.NONE); assertEquals(1, sectionsBeforeRewrite.size()); for (StreamSession.SSTableStreamingSections section : sectionsBeforeRewrite) section.ref.release(); @@ -804,7 +805,7 @@ public class SSTableRewriterTest extends SSTableWriterTestBase while (!done.get()) { Set> range = Collections.singleton(new Range(firstToken, firstToken)); - List sections = StreamSession.getSSTableSectionsForRanges(range, Collections.singleton(cfs), null); + List sections = StreamSession.getSSTableSectionsForRanges(range, Collections.singleton(cfs), null, PreviewKind.NONE); if (sections.size() != 1) failed.set(true); for (StreamSession.SSTableStreamingSections section : sections) diff --git a/test/unit/org/apache/cassandra/repair/AbstractRepairTest.java b/test/unit/org/apache/cassandra/repair/AbstractRepairTest.java index 1c508a0edb..d61d8591d4 100644 --- a/test/unit/org/apache/cassandra/repair/AbstractRepairTest.java +++ b/test/unit/org/apache/cassandra/repair/AbstractRepairTest.java @@ -34,6 +34,7 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.UUIDGen; @@ -85,7 +86,8 @@ public abstract class AbstractRepairTest Sets.newHashSet(RANGE1, RANGE2, RANGE3), isIncremental, repairedAt, - isGlobal); + isGlobal, + PreviewKind.NONE); return sessionId; } } diff --git a/test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java b/test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java index 75742dc28a..f5e9d6ba0c 100644 --- a/test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java +++ b/test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java @@ -40,6 +40,7 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.streaming.StreamPlan; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MerkleTree; import org.apache.cassandra.utils.MerkleTrees; @@ -91,7 +92,7 @@ public class LocalSyncTaskTest extends AbstractRepairTest // 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); - LocalSyncTask task = new LocalSyncTask(desc, r1, r2, null, false); + LocalSyncTask task = new LocalSyncTask(desc, r1, r2, NO_PENDING_REPAIR, false, PreviewKind.NONE); task.run(); assertEquals(0, task.get().numberOfDifferences); @@ -105,9 +106,10 @@ public class LocalSyncTaskTest extends AbstractRepairTest Keyspace keyspace = Keyspace.open(KEYSPACE1); ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Standard1"); - ActiveRepairService.instance.registerParentRepairSession(parentRepairSession, FBUtilities.getBroadcastAddress(), + ActiveRepairService.instance.registerParentRepairSession(parentRepairSession, FBUtilities.getBroadcastAddress(), Arrays.asList(cfs), Arrays.asList(range), false, - ActiveRepairService.UNREPAIRED_SSTABLE, false); + ActiveRepairService.UNREPAIRED_SSTABLE, false, + PreviewKind.NONE); RepairJobDesc desc = new RepairJobDesc(parentRepairSession, UUID.randomUUID(), KEYSPACE1, "Standard1", Arrays.asList(range)); @@ -128,7 +130,7 @@ public class LocalSyncTaskTest extends AbstractRepairTest // 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); - LocalSyncTask task = new LocalSyncTask(desc, r1, r2, null, false); + LocalSyncTask task = new LocalSyncTask(desc, r1, r2, NO_PENDING_REPAIR, false, PreviewKind.NONE); task.run(); // ensure that the changed range was recorded @@ -145,7 +147,7 @@ public class LocalSyncTaskTest extends AbstractRepairTest TreeResponse r1 = new TreeResponse(PARTICIPANT1, createInitialTree(desc, DatabaseDescriptor.getPartitioner())); TreeResponse r2 = new TreeResponse(PARTICIPANT2, createInitialTree(desc, DatabaseDescriptor.getPartitioner())); - LocalSyncTask task = new LocalSyncTask(desc, r1, r2, NO_PENDING_REPAIR, false); + LocalSyncTask task = new LocalSyncTask(desc, r1, r2, NO_PENDING_REPAIR, false, PreviewKind.NONE); StreamPlan plan = task.createStreamPlan(PARTICIPANT1, PARTICIPANT2, Lists.newArrayList(RANGE1)); assertEquals(NO_PENDING_REPAIR, plan.getPendingRepair()); @@ -162,7 +164,7 @@ public class LocalSyncTaskTest extends AbstractRepairTest TreeResponse r1 = new TreeResponse(PARTICIPANT1, createInitialTree(desc, DatabaseDescriptor.getPartitioner())); TreeResponse r2 = new TreeResponse(PARTICIPANT2, createInitialTree(desc, DatabaseDescriptor.getPartitioner())); - LocalSyncTask task = new LocalSyncTask(desc, r1, r2, desc.parentSessionId, false); + LocalSyncTask task = new LocalSyncTask(desc, r1, r2, desc.parentSessionId, false, PreviewKind.NONE); StreamPlan plan = task.createStreamPlan(PARTICIPANT1, PARTICIPANT2, Lists.newArrayList(RANGE1)); assertEquals(desc.parentSessionId, plan.getPendingRepair()); diff --git a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java index 0260cd0536..5a4e5b1448 100644 --- a/test/unit/org/apache/cassandra/repair/RepairSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/RepairSessionTest.java @@ -36,6 +36,7 @@ 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.streaming.PreviewKind; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.UUIDGen; @@ -62,7 +63,10 @@ public class RepairSessionTest IPartitioner p = Murmur3Partitioner.instance; Range repairRange = new Range<>(p.getToken(ByteBufferUtil.bytes(0)), p.getToken(ByteBufferUtil.bytes(100))); Set endpoints = Sets.newHashSet(remote); - RepairSession session = new RepairSession(parentSessionId, sessionId, Arrays.asList(repairRange), "Keyspace1", RepairParallelism.SEQUENTIAL, endpoints, false, false, "Standard1"); + RepairSession session = new RepairSession(parentSessionId, sessionId, Arrays.asList(repairRange), + "Keyspace1", RepairParallelism.SEQUENTIAL, + endpoints, false, false, + PreviewKind.NONE, "Standard1"); // perform convict session.convict(remote, Double.MAX_VALUE); diff --git a/test/unit/org/apache/cassandra/repair/StreamingRepairTaskTest.java b/test/unit/org/apache/cassandra/repair/StreamingRepairTaskTest.java index 5f13e3d26f..f433f2e42e 100644 --- a/test/unit/org/apache/cassandra/repair/StreamingRepairTaskTest.java +++ b/test/unit/org/apache/cassandra/repair/StreamingRepairTaskTest.java @@ -33,6 +33,7 @@ import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.streaming.StreamPlan; import org.apache.cassandra.utils.UUIDGen; @@ -64,8 +65,8 @@ public class StreamingRepairTaskTest extends AbstractRepairTest UUID sessionID = registerSession(cfs, true, true); ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(sessionID); RepairJobDesc desc = new RepairJobDesc(sessionID, UUIDGen.getTimeUUID(), ks, tbl, prs.getRanges()); - SyncRequest request = new SyncRequest(desc, PARTICIPANT1, PARTICIPANT2, PARTICIPANT3, prs.getRanges()); - StreamingRepairTask task = new StreamingRepairTask(desc, request, desc.sessionId); + SyncRequest request = new SyncRequest(desc, PARTICIPANT1, PARTICIPANT2, PARTICIPANT3, prs.getRanges(), PreviewKind.NONE); + StreamingRepairTask task = new StreamingRepairTask(desc, request, desc.sessionId, PreviewKind.NONE); StreamPlan plan = task.createStreamPlan(request.src, request.dst); Assert.assertFalse(plan.getFlushBeforeTransfer()); @@ -77,8 +78,8 @@ public class StreamingRepairTaskTest extends AbstractRepairTest UUID sessionID = registerSession(cfs, false, true); ActiveRepairService.ParentRepairSession prs = ActiveRepairService.instance.getParentRepairSession(sessionID); RepairJobDesc desc = new RepairJobDesc(sessionID, UUIDGen.getTimeUUID(), ks, tbl, prs.getRanges()); - SyncRequest request = new SyncRequest(desc, PARTICIPANT1, PARTICIPANT2, PARTICIPANT3, prs.getRanges()); - StreamingRepairTask task = new StreamingRepairTask(desc, request, null); + SyncRequest request = new SyncRequest(desc, PARTICIPANT1, PARTICIPANT2, PARTICIPANT3, prs.getRanges(), PreviewKind.NONE); + StreamingRepairTask task = new StreamingRepairTask(desc, request, null, PreviewKind.NONE); StreamPlan plan = task.createStreamPlan(request.src, request.dst); Assert.assertTrue(plan.getFlushBeforeTransfer()); diff --git a/test/unit/org/apache/cassandra/repair/ValidatorTest.java b/test/unit/org/apache/cassandra/repair/ValidatorTest.java index bbcdbb82cf..b45edc10ca 100644 --- a/test/unit/org/apache/cassandra/repair/ValidatorTest.java +++ b/test/unit/org/apache/cassandra/repair/ValidatorTest.java @@ -50,6 +50,7 @@ import org.apache.cassandra.repair.messages.RepairMessage; import org.apache.cassandra.repair.messages.ValidationComplete; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.MerkleTree; import org.apache.cassandra.utils.MerkleTrees; @@ -98,7 +99,7 @@ public class ValidatorTest ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(columnFamily); - Validator validator = new Validator(desc, remote, 0); + Validator validator = new Validator(desc, remote, 0, PreviewKind.NONE); MerkleTrees tree = new MerkleTrees(partitioner); tree.addMerkleTrees((int) Math.pow(2, 15), validator.desc.ranges); validator.prepare(cfs, tree); @@ -135,7 +136,7 @@ public class ValidatorTest InetAddress remote = InetAddress.getByName("127.0.0.2"); - Validator validator = new Validator(desc, remote, 0); + Validator validator = new Validator(desc, remote, 0, PreviewKind.NONE); validator.fail(); MessageOut message = outgoingMessageSink.get(TEST_TIMEOUT, TimeUnit.SECONDS); @@ -190,10 +191,10 @@ public class ValidatorTest ActiveRepairService.instance.registerParentRepairSession(repairSessionId, FBUtilities.getBroadcastAddress(), Collections.singletonList(cfs), desc.ranges, false, ActiveRepairService.UNREPAIRED_SSTABLE, - false); + false, PreviewKind.NONE); final CompletableFuture outgoingMessageSink = registerOutgoingMessageSink(); - Validator validator = new Validator(desc, FBUtilities.getBroadcastAddress(), 0, true, false); + Validator validator = new Validator(desc, FBUtilities.getBroadcastAddress(), 0, true, false, PreviewKind.NONE); CompactionManager.instance.submitValidation(cfs, validator); MessageOut message = outgoingMessageSink.get(TEST_TIMEOUT, TimeUnit.SECONDS); diff --git a/test/unit/org/apache/cassandra/repair/consistent/AbstractConsistentSessionTest.java b/test/unit/org/apache/cassandra/repair/consistent/AbstractConsistentSessionTest.java index 26168ade84..367fea990a 100644 --- a/test/unit/org/apache/cassandra/repair/consistent/AbstractConsistentSessionTest.java +++ b/test/unit/org/apache/cassandra/repair/consistent/AbstractConsistentSessionTest.java @@ -33,6 +33,7 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.UUIDGen; @@ -85,7 +86,8 @@ public abstract class AbstractConsistentSessionTest Sets.newHashSet(RANGE1, RANGE2, RANGE3), true, System.currentTimeMillis(), - true); + true, + PreviewKind.NONE); return sessionId; } } diff --git a/test/unit/org/apache/cassandra/repair/consistent/PendingAntiCompactionTest.java b/test/unit/org/apache/cassandra/repair/consistent/PendingAntiCompactionTest.java index 2cb6326faa..2126835806 100644 --- a/test/unit/org/apache/cassandra/repair/consistent/PendingAntiCompactionTest.java +++ b/test/unit/org/apache/cassandra/repair/consistent/PendingAntiCompactionTest.java @@ -52,6 +52,7 @@ import org.apache.cassandra.dht.ByteOrderedPartitioner; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.service.ActiveRepairService; import org.apache.cassandra.utils.ByteBufferUtil; @@ -146,7 +147,7 @@ public class PendingAntiCompactionTest // create a session so the anti compaction can fine it UUID sessionID = UUIDGen.getTimeUUID(); - ActiveRepairService.instance.registerParentRepairSession(sessionID, InetAddress.getLocalHost(), Lists.newArrayList(cfs), ranges, true, 1, true); + ActiveRepairService.instance.registerParentRepairSession(sessionID, InetAddress.getLocalHost(), Lists.newArrayList(cfs), ranges, true, 1, true, PreviewKind.NONE); PendingAntiCompaction pac; ExecutorService executor = Executors.newSingleThreadExecutor(); diff --git a/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java b/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java index 449a5dc4ad..5dc8112446 100644 --- a/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java +++ b/test/unit/org/apache/cassandra/repair/messages/RepairMessageSerializationsTest.java @@ -25,6 +25,7 @@ import java.util.ArrayList; import java.util.List; import java.util.UUID; +import com.google.common.collect.Lists; import org.junit.AfterClass; import org.junit.Assert; import org.junit.BeforeClass; @@ -43,10 +44,14 @@ import org.apache.cassandra.io.util.DataOutputBufferFixed; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.repair.NodePair; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.streaming.SessionSummary; +import org.apache.cassandra.streaming.StreamSummary; import org.apache.cassandra.utils.MerkleTrees; +import org.apache.cassandra.utils.UUIDGen; public class RepairMessageSerializationsTest { @@ -145,7 +150,7 @@ public class RepairMessageSerializationsTest InetAddress src = InetAddress.getByName("127.0.0.2"); InetAddress dst = InetAddress.getByName("127.0.0.3"); - SyncRequest msg = new SyncRequest(buildRepairJobDesc(), initiator, src, dst, buildTokenRanges()); + SyncRequest msg = new SyncRequest(buildRepairJobDesc(), initiator, src, dst, buildTokenRanges(), PreviewKind.NONE); serializeRoundTrip(msg, SyncRequest.serializer); } @@ -154,7 +159,12 @@ public class RepairMessageSerializationsTest { InetAddress src = InetAddress.getByName("127.0.0.2"); InetAddress dst = InetAddress.getByName("127.0.0.3"); - SyncComplete msg = new SyncComplete(buildRepairJobDesc(), new NodePair(src, dst), true); + List summaries = new ArrayList<>(); + summaries.add(new SessionSummary(src, dst, + Lists.newArrayList(new StreamSummary(TableId.fromUUID(UUIDGen.getTimeUUID()), 5, 100)), + Lists.newArrayList(new StreamSummary(TableId.fromUUID(UUIDGen.getTimeUUID()), 500, 10)) + )); + SyncComplete msg = new SyncComplete(buildRepairJobDesc(), new NodePair(src, dst), true, summaries); serializeRoundTrip(msg, SyncComplete.serializer); } @@ -162,7 +172,8 @@ public class RepairMessageSerializationsTest public void prepareMessage() throws IOException { PrepareMessage msg = new PrepareMessage(UUID.randomUUID(), new ArrayList() {{add(TableId.generate());}}, - buildTokenRanges(), true, 100000L, false); + buildTokenRanges(), true, 100000L, false, + PreviewKind.NONE); serializeRoundTrip(msg, PrepareMessage.serializer); } diff --git a/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java index 8f8fe6de7c..9d5ccf46a1 100644 --- a/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java +++ b/test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java @@ -20,7 +20,6 @@ package org.apache.cassandra.service; import java.net.InetAddress; import java.util.*; -import java.util.concurrent.ExecutionException; import com.google.common.collect.Sets; import org.junit.Before; @@ -40,6 +39,7 @@ import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.TokenMetadata; +import org.apache.cassandra.streaming.PreviewKind; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.concurrent.Refs; @@ -230,11 +230,19 @@ public class ActiveRepairServiceTest ColumnFamilyStore store = prepareColumnFamilyStore(); UUID prsId = UUID.randomUUID(); Set original = Sets.newHashSet(store.select(View.select(SSTableSet.CANONICAL, (s) -> !s.isRepaired())).sstables); - ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), store.getPartitioner().getMinimumToken())), true, System.currentTimeMillis(), true); + ActiveRepairService.instance.registerParentRepairSession(prsId, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), + Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), + store.getPartitioner().getMinimumToken())), + true, System.currentTimeMillis(), true, PreviewKind.NONE); ActiveRepairService.instance.getParentRepairSession(prsId).maybeSnapshot(store.metadata.id, prsId); UUID prsId2 = UUID.randomUUID(); - ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddress(), Collections.singletonList(store), Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), store.getPartitioner().getMinimumToken())), true, System.currentTimeMillis(), true); + ActiveRepairService.instance.registerParentRepairSession(prsId2, FBUtilities.getBroadcastAddress(), + Collections.singletonList(store), + Collections.singleton(new Range<>(store.getPartitioner().getMinimumToken(), + store.getPartitioner().getMinimumToken())), + true, System.currentTimeMillis(), + true, PreviewKind.NONE); createSSTables(store, 2); ActiveRepairService.instance.getParentRepairSession(prsId).maybeSnapshot(store.metadata.id, prsId); try (Refs refs = store.getSnapshotSSTableReaders(prsId.toString())) diff --git a/test/unit/org/apache/cassandra/service/SerializationsTest.java b/test/unit/org/apache/cassandra/service/SerializationsTest.java index 66fdeecdf1..a63dc69e49 100644 --- a/test/unit/org/apache/cassandra/service/SerializationsTest.java +++ b/test/unit/org/apache/cassandra/service/SerializationsTest.java @@ -20,10 +20,13 @@ package org.apache.cassandra.service; import java.io.IOException; import java.net.InetAddress; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.UUID; +import com.google.common.collect.Lists; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -43,8 +46,13 @@ import org.apache.cassandra.repair.NodePair; import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.Validator; import org.apache.cassandra.repair.messages.*; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.streaming.PreviewKind; +import org.apache.cassandra.streaming.SessionSummary; +import org.apache.cassandra.streaming.StreamSummary; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MerkleTrees; +import org.apache.cassandra.utils.UUIDGen; public class SerializationsTest extends AbstractSerializationsTester { @@ -115,7 +123,7 @@ public class SerializationsTest extends AbstractSerializationsTester // empty validation mt.addMerkleTree((int) Math.pow(2, 15), FULL_RANGE); - Validator v0 = new Validator(DESC, FBUtilities.getBroadcastAddress(), -1); + Validator v0 = new Validator(DESC, FBUtilities.getBroadcastAddress(), -1, PreviewKind.NONE); ValidationComplete c0 = new ValidationComplete(DESC, mt); // validation with a tree @@ -123,7 +131,7 @@ public class SerializationsTest extends AbstractSerializationsTester mt.addMerkleTree(Integer.MAX_VALUE, FULL_RANGE); for (int i = 0; i < 10; i++) mt.split(p.getRandomToken()); - Validator v1 = new Validator(DESC, FBUtilities.getBroadcastAddress(), -1); + Validator v1 = new Validator(DESC, FBUtilities.getBroadcastAddress(), -1, PreviewKind.NONE); ValidationComplete c1 = new ValidationComplete(DESC, mt); // validation failed @@ -175,8 +183,8 @@ public class SerializationsTest extends AbstractSerializationsTester InetAddress local = InetAddress.getByAddress(new byte[]{127, 0, 0, 1}); InetAddress src = InetAddress.getByAddress(new byte[]{127, 0, 0, 2}); InetAddress dest = InetAddress.getByAddress(new byte[]{127, 0, 0, 3}); - SyncRequest message = new SyncRequest(DESC, local, src, dest, Collections.singleton(FULL_RANGE)); + SyncRequest message = new SyncRequest(DESC, local, src, dest, Collections.singleton(FULL_RANGE), PreviewKind.NONE); testRepairMessageWrite("service.SyncRequest.bin", message); } @@ -209,9 +217,14 @@ public class SerializationsTest extends AbstractSerializationsTester InetAddress src = InetAddress.getByAddress(new byte[]{127, 0, 0, 2}); InetAddress dest = InetAddress.getByAddress(new byte[]{127, 0, 0, 3}); // sync success - SyncComplete success = new SyncComplete(DESC, src, dest, true); + List summaries = new ArrayList<>(); + summaries.add(new SessionSummary(src, dest, + Lists.newArrayList(new StreamSummary(TableId.fromUUID(UUIDGen.getTimeUUID()), 5, 100)), + Lists.newArrayList(new StreamSummary(TableId.fromUUID(UUIDGen.getTimeUUID()), 500, 10)) + )); + SyncComplete success = new SyncComplete(DESC, src, dest, true, summaries); // sync fail - SyncComplete fail = new SyncComplete(DESC, src, dest, false); + SyncComplete fail = new SyncComplete(DESC, src, dest, false, Collections.emptyList()); testRepairMessageWrite("service.SyncComplete.bin", success, fail); } diff --git a/test/unit/org/apache/cassandra/streaming/StreamSessionTest.java b/test/unit/org/apache/cassandra/streaming/StreamSessionTest.java index 84053d4155..6abc2a294f 100644 --- a/test/unit/org/apache/cassandra/streaming/StreamSessionTest.java +++ b/test/unit/org/apache/cassandra/streaming/StreamSessionTest.java @@ -97,7 +97,8 @@ public class StreamSessionTest Collection> ranges = Lists.newArrayList(new Range(partitioner.getMinimumToken(), partitioner.getMinimumToken())); List sections = StreamSession.getSSTableSectionsForRanges(ranges, Lists.newArrayList(cfs), - pendingRepair); + pendingRepair, + PreviewKind.NONE); Set sstables = new HashSet<>(); for (StreamSession.SSTableStreamingSections section: sections) { diff --git a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java index 57d40e91f4..21c8375713 100644 --- a/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java +++ b/test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java @@ -74,7 +74,7 @@ public class StreamTransferTaskTest public void testScheduleTimeout() throws Exception { InetAddress peer = FBUtilities.getBroadcastAddress(); - StreamSession session = new StreamSession(peer, peer, null, 0, true, null); + StreamSession session = new StreamSession(peer, peer, null, 0, true, null, PreviewKind.NONE); ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD); // create two sstables @@ -120,9 +120,9 @@ public class StreamTransferTaskTest public void testFailSessionDuringTransferShouldNotReleaseReferences() throws Exception { InetAddress peer = FBUtilities.getBroadcastAddress(); - StreamCoordinator streamCoordinator = new StreamCoordinator(1, true, null, false, null); + StreamCoordinator streamCoordinator = new StreamCoordinator(1, true, null, false, null, PreviewKind.NONE); StreamResultFuture future = StreamResultFuture.init(UUID.randomUUID(), StreamOperation.OTHER, Collections.emptyList(), streamCoordinator); - StreamSession session = new StreamSession(peer, peer, null, 0, true, null); + StreamSession session = new StreamSession(peer, peer, null, 0, true, null, PreviewKind.NONE); session.init(future); ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD);