Repository: cassandra Updated Branches: refs/heads/trunk df892a38f -> f25a765b6
Make monotonic read / read repair configurable Patch by Blake Eggleston; Reviewed by Aleksey Yeschenko for CASSANDRA-14635 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/f25a765b Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/f25a765b Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/f25a765b Branch: refs/heads/trunk Commit: f25a765b6daf4d08694303f47bfe5f38185dd7aa Parents: df892a3 Author: Blake Eggleston <[email protected]> Authored: Thu Aug 16 16:39:39 2018 -0700 Committer: Blake Eggleston <[email protected]> Committed: Fri Aug 24 09:34:52 2018 -0700 ---------------------------------------------------------------------- CHANGES.txt | 1 + doc/source/cql/ddl.rst | 32 ++ pylib/cqlshlib/cql3handling.py | 5 +- .../cql3/statements/schema/TableAttributes.java | 10 +- .../cassandra/metrics/ReadRepairMetrics.java | 2 + .../apache/cassandra/schema/SchemaKeyspace.java | 12 + .../apache/cassandra/schema/TableParams.java | 23 +- .../reads/repair/AbstractReadRepair.java | 173 +++++++++ .../reads/repair/BlockingReadRepair.java | 137 +------ .../service/reads/repair/NoopReadRepair.java | 3 + .../reads/repair/ReadOnlyReadRepair.java | 72 ++++ .../service/reads/repair/ReadRepair.java | 7 +- .../reads/repair/ReadRepairStrategy.java | 48 +++ .../cassandra/schema/SchemaKeyspaceTest.java | 11 + .../reads/repair/AbstractReadRepairTest.java | 278 ++++++++++++++ .../reads/repair/BlockingReadRepairTest.java | 307 ++++++++++++++++ .../reads/repair/InstrumentedReadRepair.java | 31 ++ .../reads/repair/ReadOnlyReadRepairTest.java | 100 +++++ .../service/reads/repair/ReadRepairTest.java | 361 ------------------- 19 files changed, 1118 insertions(+), 495 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 0e97f9a..1e8e91d 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0 + * Make monotonic read / read repair configurable (CASSANDRA-14635) * Refactor CompactionStrategyManager (CASSANDRA-14621) * Flush netty client messages immediately by default (CASSANDRA-13651) * Improve read repair blocking behavior (CASSANDRA-10726) http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/doc/source/cql/ddl.rst ---------------------------------------------------------------------- diff --git a/doc/source/cql/ddl.rst b/doc/source/cql/ddl.rst index c8cedcf..9afd638 100644 --- a/doc/source/cql/ddl.rst +++ b/doc/source/cql/ddl.rst @@ -474,6 +474,8 @@ A table supports the following options: +--------------------------------+----------+-------------+-----------------------------------------------------------+ | ``memtable_flush_period_in_ms``| *simple* | 0 | Time (in ms) before Cassandra flushes memtables to disk. | +--------------------------------+----------+-------------+-----------------------------------------------------------+ +| ``read_repair`` | *simple* | BLOCKING | Sets read repair behavior (see below) | ++--------------------------------+----------+-------------+-----------------------------------------------------------+ .. _speculative-retry-options: @@ -602,6 +604,36 @@ For instance, to create a table with both a key cache and 10 rows per partition: ) WITH caching = {'keys': 'ALL', 'rows_per_partition': 10}; +Read Repair options +################### + +The ``read_repair`` options configures the read repair behavior to allow tuning for various performance and +consistency behaviors. Two consistency properties are affected by read repair behavior. + +- Monotonic Quorum Reads: Provided by ``BLOCKING``. Monotonic quorum reads prevents reads from appearing to go back + in time in some circumstances. When monotonic quorum reads are not provided and a write fails to reach a quorum of + replicas, it may be visible in one read, and then disappear in a subsequent read. +- Write Atomicity: Provided by ``NONE``. Write atomicity prevents reads from returning partially applied writes. + Cassandra attempts to provide partition level write atomicity, but since only the data covered by a SELECT statement + is repaired by a read repair, read repair can break write atomicity when data is read at a more granular level than it + is written. For example read repair can break write atomicity if you write multiple rows to a clustered partition in a + batch, but then select a single row by specifying the clustering column in a SELECT statement. + +The available read repair settings are: + +Blocking +```````` +The default setting. When ``read_repair`` is set to ``BLOCKING``, and a read repair is triggered, the read will block +on writes sent to other replicas until the CL is reached by the writes. Provides monotonic quorum reads, but not partition +level write atomicity + +None +```` + +When ``read_repair`` is set to ``NONE``, the coordinator will reconcile any differences between replicas, but will not +attempt to repair them. Provides partition level write atomicity, but not monotonic quorum reads. + + Other considerations: ##################### http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/pylib/cqlshlib/cql3handling.py ---------------------------------------------------------------------- diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py index 5d7a3c0..5595e2a 100644 --- a/pylib/cqlshlib/cql3handling.py +++ b/pylib/cqlshlib/cql3handling.py @@ -50,7 +50,8 @@ class Cql3ParsingRuleSet(CqlParsingRuleSet): ('default_time_to_live', None), ('speculative_retry', None), ('memtable_flush_period_in_ms', None), - ('cdc', None) + ('cdc', None), + ('read_repair', None), ) columnfamily_layout_map_options = ( @@ -508,6 +509,8 @@ def cf_prop_val_completer(ctxt, cass): return [Hint('<integer>')] if this_opt in ('cdc'): return [Hint('<true|false>')] + if this_opt in ('read_repair'): + return [Hint('<\'none\'|\'blocking\'>')] return [Hint('<option_value>')] http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java b/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java index 8cc2685..c8e464a 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java +++ b/src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java @@ -25,9 +25,14 @@ import com.google.common.collect.ImmutableSet; import org.apache.cassandra.cql3.statements.PropertyDefinitions; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.exceptions.SyntaxException; -import org.apache.cassandra.schema.*; +import org.apache.cassandra.schema.CachingParams; +import org.apache.cassandra.schema.CompactionParams; +import org.apache.cassandra.schema.CompressionParams; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.schema.TableParams; import org.apache.cassandra.schema.TableParams.Option; import org.apache.cassandra.service.reads.SpeculativeRetryPolicy; +import org.apache.cassandra.service.reads.repair.ReadRepairStrategy; import static java.lang.String.format; @@ -129,6 +134,9 @@ public final class TableAttributes extends PropertyDefinitions if (hasOption(Option.CDC)) builder.cdc(getBoolean(Option.CDC.toString(), false)); + if (hasOption(Option.READ_REPAIR)) + builder.readRepair(ReadRepairStrategy.fromString(getString(Option.READ_REPAIR))); + return builder.build(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java index e639be9..fe7673d 100644 --- a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java @@ -29,6 +29,8 @@ public class ReadRepairMetrics private static final MetricNameFactory factory = new DefaultNameFactory("ReadRepair"); public static final Meter repairedBlocking = Metrics.meter(factory.createMetricName("RepairedBlocking")); + public static final Meter repairedAsync = Metrics.meter(factory.createMetricName("RepairedAsync")); + public static final Meter reconcileRead = Metrics.meter(factory.createMetricName("ReconcileRead")); @Deprecated public static final Meter repairedBackground = Metrics.meter(factory.createMetricName("RepairedBackground")); http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java index 9716862..a88aebb 100644 --- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java @@ -44,6 +44,7 @@ import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.service.reads.SpeculativeRetryPolicy; import org.apache.cassandra.schema.ColumnMetadata.ClusteringOrder; import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff; +import org.apache.cassandra.service.reads.repair.ReadRepairStrategy; import org.apache.cassandra.transport.ProtocolVersion; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; @@ -137,6 +138,7 @@ public final class SchemaKeyspace + "read_repair_chance double," // no longer used, left for drivers' sake + "speculative_retry text," + "cdc boolean," + + "read_repair text," + "PRIMARY KEY ((keyspace_name), table_name))"); private static final TableMetadata Columns = @@ -202,6 +204,7 @@ public final class SchemaKeyspace + "read_repair_chance double," // no longer used, left for drivers' sake + "speculative_retry text," + "cdc boolean," + + "read_repair text," + "PRIMARY KEY ((keyspace_name), view_name))"); private static final TableMetadata Indexes = @@ -564,6 +567,7 @@ public final class SchemaKeyspace .add("caching", params.caching.asMap()) .add("compaction", params.compaction.asMap()) .add("compression", params.compression.asMap()) + .add("read_repair", params.readRepair.toString()) .add("extensions", params.extensions); // Only add CDC-enabled flag to schema if it's enabled on the node. This is to work around RTE's post-8099 if a 3.8+ @@ -988,6 +992,7 @@ public final class SchemaKeyspace .crcCheckChance(row.getDouble("crc_check_chance")) .speculativeRetry(SpeculativeRetryPolicy.fromString(row.getString("speculative_retry"))) .cdc(row.has("cdc") && row.getBoolean("cdc")) + .readRepair(getReadRepairStrategy(row)) .build(); } @@ -1293,4 +1298,11 @@ public final class SchemaKeyspace super(message); } } + + private static ReadRepairStrategy getReadRepairStrategy(UntypedResultSet.Row row) + { + return row.has("read_repair") + ? ReadRepairStrategy.fromString(row.getString("read_repair")) + : ReadRepairStrategy.BLOCKING; + } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/schema/TableParams.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/schema/TableParams.java b/src/java/org/apache/cassandra/schema/TableParams.java index f5b3c89..afbf26c 100644 --- a/src/java/org/apache/cassandra/schema/TableParams.java +++ b/src/java/org/apache/cassandra/schema/TableParams.java @@ -28,6 +28,7 @@ import org.apache.cassandra.cql3.Attributes; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.service.reads.PercentileSpeculativeRetryPolicy; import org.apache.cassandra.service.reads.SpeculativeRetryPolicy; +import org.apache.cassandra.service.reads.repair.ReadRepairStrategy; import org.apache.cassandra.utils.BloomCalculations; import static java.lang.String.format; @@ -51,7 +52,8 @@ public final class TableParams MIN_INDEX_INTERVAL, SPECULATIVE_RETRY, CRC_CHECK_CHANCE, - CDC; + CDC, + READ_REPAIR; @Override public String toString() @@ -74,6 +76,7 @@ public final class TableParams public final CompressionParams compression; public final ImmutableMap<String, ByteBuffer> extensions; public final boolean cdc; + public final ReadRepairStrategy readRepair; private TableParams(Builder builder) { @@ -93,6 +96,7 @@ public final class TableParams compression = builder.compression; extensions = builder.extensions; cdc = builder.cdc; + readRepair = builder.readRepair; } public static Builder builder() @@ -115,7 +119,8 @@ public final class TableParams .minIndexInterval(params.minIndexInterval) .speculativeRetry(params.speculativeRetry) .extensions(params.extensions) - .cdc(params.cdc); + .cdc(params.cdc) + .readRepair(params.readRepair); } public Builder unbuild() @@ -198,7 +203,8 @@ public final class TableParams && compaction.equals(p.compaction) && compression.equals(p.compression) && extensions.equals(p.extensions) - && cdc == p.cdc; + && cdc == p.cdc + && readRepair == p.readRepair; } @Override @@ -217,7 +223,8 @@ public final class TableParams compaction, compression, extensions, - cdc); + cdc, + readRepair); } @Override @@ -238,6 +245,7 @@ public final class TableParams .add(Option.COMPRESSION.toString(), compression) .add(Option.EXTENSIONS.toString(), extensions) .add(Option.CDC.toString(), cdc) + .add(Option.READ_REPAIR.toString(), readRepair) .toString(); } @@ -257,6 +265,7 @@ public final class TableParams private CompressionParams compression = CompressionParams.DEFAULT; private ImmutableMap<String, ByteBuffer> extensions = ImmutableMap.of(); private boolean cdc; + private ReadRepairStrategy readRepair = ReadRepairStrategy.BLOCKING; public Builder() { @@ -345,6 +354,12 @@ public final class TableParams return this; } + public Builder readRepair(ReadRepairStrategy val) + { + readRepair = val; + return this; + } + public Builder extensions(Map<String, ByteBuffer> val) { extensions = ImmutableMap.copyOf(val); http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java new file mode 100644 index 0000000..a1cf827 --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java @@ -0,0 +1,173 @@ +/* + * 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.service.reads.repair; + +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; + +import com.google.common.base.Optional; +import com.google.common.base.Preconditions; +import com.google.common.collect.Iterables; +import com.google.common.collect.Sets; + +import com.codahale.metrics.Meter; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.ReadCommand; +import org.apache.cassandra.db.SinglePartitionReadCommand; +import org.apache.cassandra.db.partitions.PartitionIterator; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.ReadTimeoutException; +import org.apache.cassandra.locator.AbstractReplicationStrategy; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.NetworkTopologyStrategy; +import org.apache.cassandra.metrics.ReadRepairMetrics; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.service.reads.DataResolver; +import org.apache.cassandra.service.reads.DigestResolver; +import org.apache.cassandra.service.reads.ReadCallback; +import org.apache.cassandra.tracing.Tracing; + +public abstract class AbstractReadRepair implements ReadRepair +{ + protected final ReadCommand command; + protected final long queryStartNanoTime; + protected final ConsistencyLevel consistency; + protected final ColumnFamilyStore cfs; + + private volatile DigestRepair digestRepair = null; + + private static class DigestRepair + { + private final DataResolver dataResolver; + private final ReadCallback readCallback; + private final Consumer<PartitionIterator> resultConsumer; + private final List<InetAddressAndPort> initialContacts; + + public DigestRepair(DataResolver dataResolver, ReadCallback readCallback, Consumer<PartitionIterator> resultConsumer, List<InetAddressAndPort> initialContacts) + { + this.dataResolver = dataResolver; + this.readCallback = readCallback; + this.resultConsumer = resultConsumer; + this.initialContacts = initialContacts; + } + } + + public AbstractReadRepair(ReadCommand command, + long queryStartNanoTime, + ConsistencyLevel consistency) + { + this.command = command; + this.queryStartNanoTime = queryStartNanoTime; + this.consistency = consistency; + this.cfs = Keyspace.openAndGetStore(command.metadata()); + } + + private int getMaxResponses() + { + AbstractReplicationStrategy strategy = cfs.keyspace.getReplicationStrategy(); + if (consistency.isDatacenterLocal() && strategy instanceof NetworkTopologyStrategy) + { + NetworkTopologyStrategy nts = (NetworkTopologyStrategy) strategy; + return nts.getReplicationFactor(DatabaseDescriptor.getLocalDataCenter()); + } + else + { + return strategy.getReplicationFactor(); + } + } + + void sendReadCommand(InetAddressAndPort to, ReadCallback readCallback) + { + MessagingService.instance().sendRRWithFailure(command.createMessage(), to, readCallback); + } + + abstract Meter getRepairMeter(); + + // digestResolver isn't used here because we resend read requests to all participants + public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer) + { + getRepairMeter().mark(); + + // Do a full data read to resolve the correct response (and repair node that need be) + Keyspace keyspace = Keyspace.open(command.metadata().keyspace); + DataResolver resolver = new DataResolver(keyspace, command, ConsistencyLevel.ALL, getMaxResponses(), queryStartNanoTime, this); + ReadCallback readCallback = new ReadCallback(resolver, ConsistencyLevel.ALL, consistency.blockFor(cfs.keyspace), command, + keyspace, allEndpoints, queryStartNanoTime); + + digestRepair = new DigestRepair(resolver, readCallback, resultConsumer, contactedEndpoints); + + for (InetAddressAndPort endpoint : contactedEndpoints) + { + Tracing.trace("Enqueuing full data read to {}", endpoint); + sendReadCommand(endpoint, readCallback); + } + } + + public void awaitReads() throws ReadTimeoutException + { + DigestRepair repair = digestRepair; + if (repair == null) + return; + + repair.readCallback.awaitResults(); + repair.resultConsumer.accept(digestRepair.dataResolver.resolve()); + } + + private boolean shouldSpeculate() + { + ConsistencyLevel speculativeCL = consistency.isDatacenterLocal() ? ConsistencyLevel.LOCAL_QUORUM : ConsistencyLevel.QUORUM; + return consistency != ConsistencyLevel.EACH_QUORUM + && consistency.satisfies(speculativeCL, cfs.keyspace) + && cfs.sampleLatencyNanos <= TimeUnit.MILLISECONDS.toNanos(command.getTimeout()); + } + + Iterable<InetAddressAndPort> getCandidatesForToken(Token token) + { + return BlockingReadRepairs.getCandidateEndpoints(cfs.keyspace, token, consistency); + } + + public void maybeSendAdditionalReads() + { + Preconditions.checkState(command instanceof SinglePartitionReadCommand, + "maybeSendAdditionalReads can only be called for SinglePartitionReadCommand"); + DigestRepair repair = digestRepair; + if (repair == null) + return; + + if (shouldSpeculate() && !repair.readCallback.await(cfs.sampleLatencyNanos, TimeUnit.NANOSECONDS)) + { + Set<InetAddressAndPort> contacted = Sets.newHashSet(repair.initialContacts); + Token replicaToken = ((SinglePartitionReadCommand) command).partitionKey().getToken(); + Iterable<InetAddressAndPort> candidates = getCandidatesForToken(replicaToken); + + Optional<InetAddressAndPort> endpoint = Iterables.tryFind(candidates, e -> !contacted.contains(e)); + if (endpoint.isPresent()) + { + Tracing.trace("Enqueuing speculative full data read to {}", endpoint); + sendReadCommand(endpoint.get(), repair.readCallback); + ReadRepairMetrics.speculatedRead.mark(); + } + } + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java index c5f1bea..e46372e 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java @@ -18,84 +18,40 @@ package org.apache.cassandra.service.reads.repair; -import java.util.List; import java.util.Map; import java.util.Queue; -import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; -import java.util.function.Consumer; - -import com.google.common.base.Preconditions; -import com.google.common.collect.Iterables; -import com.google.common.collect.Sets; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.codahale.metrics.Meter; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.DecoratedKey; -import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.ReadCommand; -import org.apache.cassandra.db.SinglePartitionReadCommand; -import org.apache.cassandra.db.partitions.PartitionIterator; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; -import org.apache.cassandra.dht.Token; import org.apache.cassandra.exceptions.ReadTimeoutException; -import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.locator.NetworkTopologyStrategy; import org.apache.cassandra.metrics.ReadRepairMetrics; -import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.service.reads.DataResolver; -import org.apache.cassandra.service.reads.DigestResolver; -import org.apache.cassandra.service.reads.ReadCallback; import org.apache.cassandra.tracing.Tracing; /** * 'Classic' read repair. Doesn't allow the client read to return until - * updates have been written to nodes needing correction. + * updates have been written to nodes needing correction. Breaks write + * atomicity in some situations */ -public class BlockingReadRepair implements ReadRepair +public class BlockingReadRepair extends AbstractReadRepair { private static final Logger logger = LoggerFactory.getLogger(BlockingReadRepair.class); - private final ReadCommand command; - private final long queryStartNanoTime; - private final ConsistencyLevel consistency; - private final ColumnFamilyStore cfs; - - private final Queue<BlockingPartitionRepair> repairs = new ConcurrentLinkedQueue<>(); - - private volatile DigestRepair digestRepair = null; - - private static class DigestRepair - { - private final DataResolver dataResolver; - private final ReadCallback readCallback; - private final Consumer<PartitionIterator> resultConsumer; - private final List<InetAddressAndPort> initialContacts; - - public DigestRepair(DataResolver dataResolver, ReadCallback readCallback, Consumer<PartitionIterator> resultConsumer, List<InetAddressAndPort> initialContacts) - { - this.dataResolver = dataResolver; - this.readCallback = readCallback; - this.resultConsumer = resultConsumer; - this.initialContacts = initialContacts; - } - } + protected final Queue<BlockingPartitionRepair> repairs = new ConcurrentLinkedQueue<>(); - public BlockingReadRepair(ReadCommand command, - long queryStartNanoTime, - ConsistencyLevel consistency) + public BlockingReadRepair(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) { - this.command = command; - this.queryStartNanoTime = queryStartNanoTime; - this.consistency = consistency; - this.cfs = Keyspace.openAndGetStore(command.metadata()); + super(command, queryStartNanoTime, consistency); } public UnfilteredPartitionIterators.MergeListener getMergeListener(InetAddressAndPort[] endpoints) @@ -103,83 +59,10 @@ public class BlockingReadRepair implements ReadRepair return new PartitionIteratorMergeListener(endpoints, command, consistency, this); } - private int getMaxResponses() - { - AbstractReplicationStrategy strategy = cfs.keyspace.getReplicationStrategy(); - if (consistency.isDatacenterLocal() && strategy instanceof NetworkTopologyStrategy) - { - NetworkTopologyStrategy nts = (NetworkTopologyStrategy) strategy; - return nts.getReplicationFactor(DatabaseDescriptor.getLocalDataCenter()); - } - else - { - return strategy.getReplicationFactor(); - } - } - - // digestResolver isn't used here because we resend read requests to all participants - public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer) - { - ReadRepairMetrics.repairedBlocking.mark(); - - // Do a full data read to resolve the correct response (and repair node that need be) - Keyspace keyspace = Keyspace.open(command.metadata().keyspace); - DataResolver resolver = new DataResolver(keyspace, command, ConsistencyLevel.ALL, getMaxResponses(), queryStartNanoTime, this); - ReadCallback readCallback = new ReadCallback(resolver, ConsistencyLevel.ALL, consistency.blockFor(cfs.keyspace), command, - keyspace, allEndpoints, queryStartNanoTime); - - digestRepair = new DigestRepair(resolver, readCallback, resultConsumer, contactedEndpoints); - - for (InetAddressAndPort endpoint : contactedEndpoints) - { - Tracing.trace("Enqueuing full data read to {}", endpoint); - MessagingService.instance().sendRRWithFailure(command.createMessage(), endpoint, readCallback); - } - } - - public void awaitReads() throws ReadTimeoutException - { - DigestRepair repair = digestRepair; - if (repair == null) - return; - - repair.readCallback.awaitResults(); - repair.resultConsumer.accept(digestRepair.dataResolver.resolve()); - } - - private boolean shouldSpeculate() - { - ConsistencyLevel speculativeCL = consistency.isDatacenterLocal() ? ConsistencyLevel.LOCAL_QUORUM : ConsistencyLevel.QUORUM; - return consistency != ConsistencyLevel.EACH_QUORUM - && consistency.satisfies(speculativeCL, cfs.keyspace) - && cfs.sampleLatencyNanos <= TimeUnit.MILLISECONDS.toNanos(command.getTimeout()); - } - - public void maybeSendAdditionalReads() + @Override + Meter getRepairMeter() { - Preconditions.checkState(command instanceof SinglePartitionReadCommand, - "maybeSendAdditionalReads can only be called for SinglePartitionReadCommand"); - DigestRepair repair = digestRepair; - if (repair == null) - return; - - if (shouldSpeculate() && !repair.readCallback.await(cfs.sampleLatencyNanos, TimeUnit.NANOSECONDS)) - { - Set<InetAddressAndPort> contacted = Sets.newHashSet(repair.initialContacts); - Token replicaToken = ((SinglePartitionReadCommand) command).partitionKey().getToken(); - Iterable<InetAddressAndPort> candidates = BlockingReadRepairs.getCandidateEndpoints(cfs.keyspace, replicaToken, consistency); - boolean speculated = false; - for (InetAddressAndPort endpoint: Iterables.filter(candidates, e -> !contacted.contains(e))) - { - speculated = true; - Tracing.trace("Enqueuing speculative full data read to {}", endpoint); - MessagingService.instance().sendRR(command.createMessage(), endpoint, repair.readCallback); - break; - } - - if (speculated) - ReadRepairMetrics.speculatedRead.mark(); - } + return ReadRepairMetrics.repairedBlocking; } @Override http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java index 6e161a8..a43e3eb 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java @@ -30,6 +30,9 @@ import org.apache.cassandra.exceptions.ReadTimeoutException; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.service.reads.DigestResolver; +/** + * Bypasses the read repair path for short read protection and testing + */ public class NoopReadRepair implements ReadRepair { public static final NoopReadRepair instance = new NoopReadRepair(); http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/service/reads/repair/ReadOnlyReadRepair.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/ReadOnlyReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/ReadOnlyReadRepair.java new file mode 100644 index 0000000..d994b23 --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/repair/ReadOnlyReadRepair.java @@ -0,0 +1,72 @@ +/* + * 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.service.reads.repair; + +import java.util.Map; + +import com.codahale.metrics.Meter; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.ReadCommand; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.ReadRepairMetrics; + +/** + * Only performs the collection of data responses and reconciliation of them, doesn't send repair mutations + * to replicas. This preserves write atomicity, but doesn't provide monotonic quorum reads + */ +public class ReadOnlyReadRepair extends AbstractReadRepair +{ + public ReadOnlyReadRepair(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) + { + super(command, queryStartNanoTime, consistency); + } + + @Override + public UnfilteredPartitionIterators.MergeListener getMergeListener(InetAddressAndPort[] endpoints) + { + return UnfilteredPartitionIterators.MergeListener.NOOP; + } + + @Override + Meter getRepairMeter() + { + return ReadRepairMetrics.reconcileRead; + } + + @Override + public void maybeSendAdditionalWrites() + { + + } + + @Override + public void repairPartition(DecoratedKey key, Map<InetAddressAndPort, Mutation> mutations, InetAddressAndPort[] destinations) + { + throw new UnsupportedOperationException("ReadOnlyReadRepair shouldn't be trying to repair partitions"); + } + + @Override + public void awaitWrites() + { + + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java index a1a9546..97f0f67 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java @@ -33,6 +33,11 @@ import org.apache.cassandra.service.reads.DigestResolver; public interface ReadRepair { + public interface Factory + { + ReadRepair create(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency); + } + /** * Used by DataResolver to generate corrections as the partition iterator is consumed */ @@ -87,6 +92,6 @@ public interface ReadRepair static ReadRepair create(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) { - return new BlockingReadRepair(command, queryStartNanoTime, consistency); + return command.metadata().params.readRepair.create(command, queryStartNanoTime, consistency); } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/src/java/org/apache/cassandra/service/reads/repair/ReadRepairStrategy.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/ReadRepairStrategy.java b/src/java/org/apache/cassandra/service/reads/repair/ReadRepairStrategy.java new file mode 100644 index 0000000..5945633 --- /dev/null +++ b/src/java/org/apache/cassandra/service/reads/repair/ReadRepairStrategy.java @@ -0,0 +1,48 @@ +/* + * 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.service.reads.repair; + +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.ReadCommand; + +public enum ReadRepairStrategy implements ReadRepair.Factory +{ + NONE + { + @Override + public ReadRepair create(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) + { + return new ReadOnlyReadRepair(command, queryStartNanoTime, consistency); + } + }, + + BLOCKING + { + @Override + public ReadRepair create(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) + { + return new BlockingReadRepair(command, queryStartNanoTime, consistency); + } + }; + + public static ReadRepairStrategy fromString(String s) + { + return valueOf(s.toUpperCase()); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/test/unit/org/apache/cassandra/schema/SchemaKeyspaceTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/schema/SchemaKeyspaceTest.java b/test/unit/org/apache/cassandra/schema/SchemaKeyspaceTest.java index 501c429..bf62203 100644 --- a/test/unit/org/apache/cassandra/schema/SchemaKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/schema/SchemaKeyspaceTest.java @@ -26,6 +26,7 @@ import java.util.Set; import com.google.common.collect.ImmutableMap; +import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; @@ -39,6 +40,7 @@ import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.partitions.PartitionUpdate; import org.apache.cassandra.db.rows.UnfilteredRowIterators; import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.service.reads.repair.ReadRepairStrategy; import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal; @@ -96,6 +98,15 @@ public class SchemaKeyspaceTest assertEquals(extensions, metadata.params.extensions); } + @Test + public void testReadRepair() + { + createTable("ks", "CREATE TABLE tbl (a text primary key, b int, c int) WITH read_repair='none'"); + TableMetadata metadata = Schema.instance.getTableMetadata("ks", "tbl"); + Assert.assertEquals(ReadRepairStrategy.NONE, metadata.params.readRepair); + + } + private static void updateTable(String keyspace, TableMetadata oldTable, TableMetadata newTable) { KeyspaceMetadata ksm = Schema.instance.getKeyspaceInstance(keyspace).getMetadata(); http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java b/test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java new file mode 100644 index 0000000..9717c4e --- /dev/null +++ b/test/unit/org/apache/cassandra/service/reads/repair/AbstractReadRepairTest.java @@ -0,0 +1,278 @@ +package org.apache.cassandra.service.reads.repair; + +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Iterables; +import com.google.common.collect.Lists; +import com.google.common.collect.Sets; +import com.google.common.primitives.Ints; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.Util; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.cql3.ColumnIdentifier; +import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; +import org.apache.cassandra.db.Clustering; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.ReadCommand; +import org.apache.cassandra.db.ReadResponse; +import org.apache.cassandra.db.partitions.PartitionIterator; +import org.apache.cassandra.db.partitions.PartitionUpdate; +import org.apache.cassandra.db.partitions.SingletonUnfilteredPartitionIterator; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; +import org.apache.cassandra.db.rows.BTreeRow; +import org.apache.cassandra.db.rows.BufferCell; +import org.apache.cassandra.db.rows.Cell; +import org.apache.cassandra.db.rows.Row; +import org.apache.cassandra.db.rows.RowIterator; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessageIn; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.schema.MigrationManager; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.schema.Tables; +import org.apache.cassandra.utils.ByteBufferUtil; + +@Ignore +public abstract class AbstractReadRepairTest +{ + static Keyspace ks; + static ColumnFamilyStore cfs; + static TableMetadata cfm; + static InetAddressAndPort target1; + static InetAddressAndPort target2; + static InetAddressAndPort target3; + static List<InetAddressAndPort> targets; + + static long now = TimeUnit.NANOSECONDS.toMicros(System.nanoTime()); + static DecoratedKey key; + static Cell cell1; + static Cell cell2; + static Cell cell3; + static Mutation resolved; + + static ReadCommand command; + + static void assertRowsEqual(Row expected, Row actual) + { + try + { + Assert.assertEquals(expected == null, actual == null); + if (expected == null) + return; + Assert.assertEquals(expected.clustering(), actual.clustering()); + Assert.assertEquals(expected.deletion(), actual.deletion()); + Assert.assertArrayEquals(Iterables.toArray(expected.cells(), Cell.class), Iterables.toArray(expected.cells(), Cell.class)); + } catch (Throwable t) + { + throw new AssertionError(String.format("Row comparison failed, expected %s got %s", expected, actual), t); + } + } + + static void assertRowsEqual(RowIterator expected, RowIterator actual) + { + assertRowsEqual(expected.staticRow(), actual.staticRow()); + while (expected.hasNext()) + { + assert actual.hasNext(); + assertRowsEqual(expected.next(), actual.next()); + } + assert !actual.hasNext(); + } + + static void assertPartitionsEqual(PartitionIterator expected, PartitionIterator actual) + { + while (expected.hasNext()) + { + assert actual.hasNext(); + assertRowsEqual(expected.next(), actual.next()); + } + + assert !actual.hasNext(); + } + + static void assertMutationEqual(Mutation expected, Mutation actual) + { + Assert.assertEquals(expected.getKeyspaceName(), actual.getKeyspaceName()); + Assert.assertEquals(expected.key(), actual.key()); + Assert.assertEquals(expected.key(), actual.key()); + PartitionUpdate expectedUpdate = Iterables.getOnlyElement(expected.getPartitionUpdates()); + PartitionUpdate actualUpdate = Iterables.getOnlyElement(actual.getPartitionUpdates()); + assertRowsEqual(Iterables.getOnlyElement(expectedUpdate), Iterables.getOnlyElement(actualUpdate)); + } + + static DecoratedKey dk(int v) + { + return DatabaseDescriptor.getPartitioner().decorateKey(ByteBufferUtil.bytes(v)); + } + + static Cell cell(String name, String value, long timestamp) + { + return BufferCell.live(cfm.getColumn(ColumnIdentifier.getInterned(name, false)), timestamp, ByteBufferUtil.bytes(value)); + } + + static PartitionUpdate update(Cell... cells) + { + Row.Builder builder = BTreeRow.unsortedBuilder(); + builder.newRow(Clustering.EMPTY); + for (Cell cell: cells) + { + builder.addCell(cell); + } + return PartitionUpdate.singleRowUpdate(cfm, key, builder.build()); + } + + static PartitionIterator partition(Cell... cells) + { + UnfilteredPartitionIterator iter = new SingletonUnfilteredPartitionIterator(update(cells).unfilteredIterator()); + return UnfilteredPartitionIterators.filter(iter, Ints.checkedCast(TimeUnit.MICROSECONDS.toSeconds(now))); + } + + static Mutation mutation(Cell... cells) + { + return new Mutation(update(cells)); + } + + @SuppressWarnings("resource") + static MessageIn<ReadResponse> msg(InetAddressAndPort from, Cell... cells) + { + UnfilteredPartitionIterator iter = new SingletonUnfilteredPartitionIterator(update(cells).unfilteredIterator()); + return MessageIn.create(from, + ReadResponse.createDataResponse(iter, command), + Collections.emptyMap(), + MessagingService.Verb.INTERNAL_RESPONSE, + MessagingService.current_version); + } + + static class ResultConsumer implements Consumer<PartitionIterator> + { + + PartitionIterator result = null; + + @Override + public void accept(PartitionIterator partitionIterator) + { + Assert.assertNotNull(partitionIterator); + result = partitionIterator; + } + } + + private static boolean configured = false; + + static void configureClass(ReadRepairStrategy repairStrategy) throws Throwable + { + SchemaLoader.loadSchema(); + String ksName = "ks"; + + String ddl = String.format("CREATE TABLE tbl (k int primary key, v text) WITH read_repair='%s'", + repairStrategy.toString().toLowerCase()); + + cfm = CreateTableStatement.parse(ddl, ksName).build(); + assert cfm.params.readRepair == repairStrategy; + KeyspaceMetadata ksm = KeyspaceMetadata.create(ksName, KeyspaceParams.simple(3), Tables.of(cfm)); + MigrationManager.announceNewKeyspace(ksm, false); + + ks = Keyspace.open(ksName); + cfs = ks.getColumnFamilyStore("tbl"); + + cfs.sampleLatencyNanos = 0; + + target1 = InetAddressAndPort.getByName("127.0.0.255"); + target2 = InetAddressAndPort.getByName("127.0.0.254"); + target3 = InetAddressAndPort.getByName("127.0.0.253"); + + targets = ImmutableList.of(target1, target2, target3); + + // default test values + key = dk(5); + cell1 = cell("v", "val1", now); + cell2 = cell("v", "val2", now); + cell3 = cell("v", "val3", now); + resolved = mutation(cell1, cell2); + + command = Util.cmd(cfs, 1).build(); + + configured = true; + } + + static Set<InetAddressAndPort> epSet(InetAddressAndPort... eps) + { + return Sets.newHashSet(eps); + } + + @Before + public void setUp() + { + assert configured : "configureClass must be called in a @BeforeClass method"; + cfs.sampleLatencyNanos = 0; + } + + public abstract InstrumentedReadRepair createInstrumentedReadRepair(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency); + + public InstrumentedReadRepair createInstrumentedReadRepair() + { + return createInstrumentedReadRepair(command, System.nanoTime(), ConsistencyLevel.QUORUM); + + } + + /** + * If we haven't received enough full data responses by the time the speculation + * timeout occurs, we should send read requests to additional replicas + */ + @Test + public void readSpeculationCycle() + { + InstrumentedReadRepair repair = createInstrumentedReadRepair(); + ResultConsumer consumer = new ResultConsumer(); + + + Assert.assertEquals(epSet(), repair.getReadRecipients()); + repair.startRepair(null, targets, Lists.newArrayList(target1, target2), consumer); + + Assert.assertEquals(epSet(target1, target2), repair.getReadRecipients()); + repair.maybeSendAdditionalReads(); + Assert.assertEquals(epSet(target1, target2, target3), repair.getReadRecipients()); + Assert.assertNull(consumer.result); + } + + /** + * If we receive enough data responses by the before the speculation timeout + * passes, we shouldn't send additional read requests + */ + @Test + public void noSpeculationRequired() + { + InstrumentedReadRepair repair = createInstrumentedReadRepair(); + ResultConsumer consumer = new ResultConsumer(); + + Assert.assertEquals(epSet(), repair.getReadRecipients()); + repair.startRepair(null, targets, Lists.newArrayList(target1, target2), consumer); + + Assert.assertEquals(epSet(target1, target2), repair.getReadRecipients()); + repair.getReadCallback().response(msg(target1, cell1)); + repair.getReadCallback().response(msg(target2, cell1)); + + repair.maybeSendAdditionalReads(); + Assert.assertEquals(epSet(target1, target2), repair.getReadRecipients()); + + repair.awaitReads(); + + assertPartitionsEqual(partition(cell1), consumer.result); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/test/unit/org/apache/cassandra/service/reads/repair/BlockingReadRepairTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/service/reads/repair/BlockingReadRepairTest.java b/test/unit/org/apache/cassandra/service/reads/repair/BlockingReadRepairTest.java new file mode 100644 index 0000000..b06e88a --- /dev/null +++ b/test/unit/org/apache/cassandra/service/reads/repair/BlockingReadRepairTest.java @@ -0,0 +1,307 @@ +/* + * 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.service.reads.repair; + +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import com.google.common.collect.Lists; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.ReadCommand; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.MessageOut; +import org.apache.cassandra.service.reads.ReadCallback; + +public class BlockingReadRepairTest extends AbstractReadRepairTest +{ + + private static class InstrumentedReadRepairHandler extends BlockingPartitionRepair + { + public InstrumentedReadRepairHandler(Keyspace keyspace, DecoratedKey key, ConsistencyLevel consistency, Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor, InetAddressAndPort[] participants) + { + super(keyspace, key, consistency, repairs, maxBlockFor, participants); + } + + Map<InetAddressAndPort, Mutation> mutationsSent = new HashMap<>(); + + protected void sendRR(MessageOut<Mutation> message, InetAddressAndPort endpoint) + { + mutationsSent.put(endpoint, message.payload); + } + + List<InetAddressAndPort> candidates = targets; + + protected List<InetAddressAndPort> getCandidateEndpoints() + { + return candidates; + } + + @Override + protected boolean isLocal(InetAddressAndPort endpoint) + { + return targets.contains(endpoint); + } + } + + @BeforeClass + public static void setUpClass() throws Throwable + { + configureClass(ReadRepairStrategy.BLOCKING); + } + + private static InstrumentedReadRepairHandler createRepairHandler(Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor, Collection<InetAddressAndPort> participants) + { + InetAddressAndPort[] participantArray = new InetAddressAndPort[participants.size()]; + participants.toArray(participantArray); + return new InstrumentedReadRepairHandler(ks, key, ConsistencyLevel.LOCAL_QUORUM, repairs, maxBlockFor, participantArray); + } + + private static InstrumentedReadRepairHandler createRepairHandler(Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor) + { + return createRepairHandler(repairs, maxBlockFor, repairs.keySet()); + } + + private static class InstrumentedBlockingReadRepair extends BlockingReadRepair implements InstrumentedReadRepair + { + public InstrumentedBlockingReadRepair(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) + { + super(command, queryStartNanoTime, consistency); + } + + Set<InetAddressAndPort> readCommandRecipients = new HashSet<>(); + ReadCallback readCallback = null; + + @Override + void sendReadCommand(InetAddressAndPort to, ReadCallback callback) + { + assert readCallback == null || readCallback == callback; + readCommandRecipients.add(to); + readCallback = callback; + } + + @Override + Iterable<InetAddressAndPort> getCandidatesForToken(Token token) + { + return targets; + } + + @Override + public Set<InetAddressAndPort> getReadRecipients() + { + return readCommandRecipients; + } + + @Override + public ReadCallback getReadCallback() + { + return readCallback; + } + } + + @Override + public InstrumentedReadRepair createInstrumentedReadRepair(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) + { + return new InstrumentedBlockingReadRepair(command, queryStartNanoTime, consistency); + } + + @Test + public void consistencyLevelTest() throws Exception + { + Assert.assertTrue(ConsistencyLevel.QUORUM.satisfies(ConsistencyLevel.QUORUM, ks)); + Assert.assertTrue(ConsistencyLevel.THREE.satisfies(ConsistencyLevel.QUORUM, ks)); + Assert.assertTrue(ConsistencyLevel.TWO.satisfies(ConsistencyLevel.QUORUM, ks)); + Assert.assertFalse(ConsistencyLevel.ONE.satisfies(ConsistencyLevel.QUORUM, ks)); + Assert.assertFalse(ConsistencyLevel.ANY.satisfies(ConsistencyLevel.QUORUM, ks)); + } + + + @Test + public void additionalMutationRequired() throws Exception + { + + Mutation repair1 = mutation(cell2); + Mutation repair2 = mutation(cell1); + + // check that the correct repairs are calculated + Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); + repairs.put(target1, repair1); + repairs.put(target2, repair2); + + + InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2); + + Assert.assertTrue(handler.mutationsSent.isEmpty()); + + // check that the correct mutations are sent + handler.sendInitialRepairs(); + Assert.assertEquals(2, handler.mutationsSent.size()); + assertMutationEqual(repair1, handler.mutationsSent.get(target1)); + assertMutationEqual(repair2, handler.mutationsSent.get(target2)); + + // check that a combined mutation is speculatively sent to the 3rd target + handler.mutationsSent.clear(); + handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS); + Assert.assertEquals(1, handler.mutationsSent.size()); + assertMutationEqual(resolved, handler.mutationsSent.get(target3)); + + // check repairs stop blocking after receiving 2 acks + Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + handler.ack(target1); + Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + handler.ack(target3); + Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + + } + + /** + * If we've received enough acks, we shouldn't send any additional mutations + */ + @Test + public void noAdditionalMutationRequired() throws Exception + { + Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); + repairs.put(target1, mutation(cell2)); + repairs.put(target2, mutation(cell1)); + + InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2); + handler.sendInitialRepairs(); + handler.ack(target1); + handler.ack(target2); + + // both replicas have acked, we shouldn't send anything else out + handler.mutationsSent.clear(); + handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS); + Assert.assertTrue(handler.mutationsSent.isEmpty()); + } + + /** + * If there are no additional nodes we can send mutations to, we... shouldn't + */ + @Test + public void noAdditionalMutationPossible() throws Exception + { + Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); + repairs.put(target1, mutation(cell2)); + repairs.put(target2, mutation(cell1)); + + InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2); + handler.sendInitialRepairs(); + + // we've already sent mutations to all candidates, so we shouldn't send any more + handler.candidates = Lists.newArrayList(target1, target2); + handler.mutationsSent.clear(); + handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS); + Assert.assertTrue(handler.mutationsSent.isEmpty()); + } + + /** + * If we didn't send a repair to a replica because there wasn't a diff with the + * resolved column family, we shouldn't send it a speculative mutation + */ + @Test + public void mutationsArentSentToInSyncNodes() throws Exception + { + Mutation repair1 = mutation(cell2); + + Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); + repairs.put(target1, repair1); + Collection<InetAddressAndPort> participants = Lists.newArrayList(target1, target2); + + // check that the correct initial mutations are sent out + InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2, participants); + handler.sendInitialRepairs(); + Assert.assertEquals(1, handler.mutationsSent.size()); + Assert.assertTrue(handler.mutationsSent.containsKey(target1)); + + // check that speculative mutations aren't sent to target2 + handler.mutationsSent.clear(); + handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS); + Assert.assertEquals(1, handler.mutationsSent.size()); + Assert.assertTrue(handler.mutationsSent.containsKey(target3)); + } + + @Test + public void onlyBlockOnQuorum() + { + Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); + repairs.put(target1, mutation(cell1)); + repairs.put(target2, mutation(cell2)); + repairs.put(target3, mutation(cell3)); + Assert.assertEquals(3, repairs.size()); + + InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2); + handler.sendInitialRepairs(); + + Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + handler.ack(target1); + Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + + // here we should stop blocking, even though we've sent 3 repairs + handler.ack(target2); + Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + + } + + /** + * For dc local consistency levels, noop mutations and responses from remote dcs should not affect effective blockFor + */ + @Test + public void remoteDCTest() throws Exception + { + Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); + repairs.put(target1, mutation(cell1)); + + + InetAddressAndPort remote1 = InetAddressAndPort.getByName("10.0.0.1"); + InetAddressAndPort remote2 = InetAddressAndPort.getByName("10.0.0.2"); + repairs.put(remote1, mutation(cell1)); + + Collection<InetAddressAndPort> participants = Lists.newArrayList(target1, target2, remote1, remote2); + + InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2, participants); + handler.sendInitialRepairs(); + Assert.assertEquals(2, handler.mutationsSent.size()); + Assert.assertTrue(handler.mutationsSent.containsKey(target1)); + Assert.assertTrue(handler.mutationsSent.containsKey(remote1)); + + Assert.assertEquals(1, handler.waitingOn()); + Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + + handler.ack(remote1); + Assert.assertEquals(1, handler.waitingOn()); + Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + + handler.ack(target1); + Assert.assertEquals(0, handler.waitingOn()); + Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/test/unit/org/apache/cassandra/service/reads/repair/InstrumentedReadRepair.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/service/reads/repair/InstrumentedReadRepair.java b/test/unit/org/apache/cassandra/service/reads/repair/InstrumentedReadRepair.java new file mode 100644 index 0000000..2fb8ffc --- /dev/null +++ b/test/unit/org/apache/cassandra/service/reads/repair/InstrumentedReadRepair.java @@ -0,0 +1,31 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.service.reads.repair; + +import java.util.Set; + +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.service.reads.ReadCallback; + +public interface InstrumentedReadRepair extends ReadRepair +{ + Set<InetAddressAndPort> getReadRecipients(); + + ReadCallback getReadCallback(); +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/test/unit/org/apache/cassandra/service/reads/repair/ReadOnlyReadRepairTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/service/reads/repair/ReadOnlyReadRepairTest.java b/test/unit/org/apache/cassandra/service/reads/repair/ReadOnlyReadRepairTest.java new file mode 100644 index 0000000..efce59a --- /dev/null +++ b/test/unit/org/apache/cassandra/service/reads/repair/ReadOnlyReadRepairTest.java @@ -0,0 +1,100 @@ +/* + * 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.service.reads.repair; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.db.ConsistencyLevel; +import org.apache.cassandra.db.ReadCommand; +import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.service.reads.ReadCallback; + +public class ReadOnlyReadRepairTest extends AbstractReadRepairTest +{ + private static class InstrumentedReadOnlyReadRepair extends ReadOnlyReadRepair implements InstrumentedReadRepair + { + public InstrumentedReadOnlyReadRepair(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) + { + super(command, queryStartNanoTime, consistency); + } + + Set<InetAddressAndPort> readCommandRecipients = new HashSet<>(); + ReadCallback readCallback = null; + + @Override + void sendReadCommand(InetAddressAndPort to, ReadCallback callback) + { + assert readCallback == null || readCallback == callback; + readCommandRecipients.add(to); + readCallback = callback; + } + + @Override + Iterable<InetAddressAndPort> getCandidatesForToken(Token token) + { + return targets; + } + + @Override + public Set<InetAddressAndPort> getReadRecipients() + { + return readCommandRecipients; + } + + @Override + public ReadCallback getReadCallback() + { + return readCallback; + } + } + + @BeforeClass + public static void setUpClass() throws Throwable + { + configureClass(ReadRepairStrategy.NONE); + } + + @Override + public InstrumentedReadRepair createInstrumentedReadRepair(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency) + { + return new InstrumentedReadOnlyReadRepair(command, queryStartNanoTime, consistency); + } + + @Test + public void getMergeListener() + { + InstrumentedReadRepair repair = createInstrumentedReadRepair(); + Assert.assertSame(UnfilteredPartitionIterators.MergeListener.NOOP, repair.getMergeListener(new InetAddressAndPort[]{})); + } + + @Test(expected = UnsupportedOperationException.class) + public void repairPartitionFailure() + { + InstrumentedReadRepair repair = createInstrumentedReadRepair(); + repair.repairPartition(dk(1), Collections.emptyMap(), new InetAddressAndPort[]{}); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/f25a765b/test/unit/org/apache/cassandra/service/reads/repair/ReadRepairTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/service/reads/repair/ReadRepairTest.java b/test/unit/org/apache/cassandra/service/reads/repair/ReadRepairTest.java deleted file mode 100644 index 9f06bb2..0000000 --- a/test/unit/org/apache/cassandra/service/reads/repair/ReadRepairTest.java +++ /dev/null @@ -1,361 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.cassandra.service.reads.repair; - -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeUnit; - -import com.google.common.collect.ImmutableList; -import com.google.common.collect.Iterables; -import com.google.common.collect.Lists; -import org.junit.Assert; -import org.junit.BeforeClass; -import org.junit.Test; - -import org.apache.cassandra.SchemaLoader; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.cql3.ColumnIdentifier; -import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; -import org.apache.cassandra.db.Clustering; -import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.db.ConsistencyLevel; -import org.apache.cassandra.db.DecoratedKey; -import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.db.Mutation; -import org.apache.cassandra.db.partitions.PartitionUpdate; -import org.apache.cassandra.db.rows.BTreeRow; -import org.apache.cassandra.db.rows.BufferCell; -import org.apache.cassandra.db.rows.Cell; -import org.apache.cassandra.db.rows.Row; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.net.MessageOut; -import org.apache.cassandra.schema.KeyspaceMetadata; -import org.apache.cassandra.schema.KeyspaceParams; -import org.apache.cassandra.schema.MigrationManager; -import org.apache.cassandra.schema.TableMetadata; -import org.apache.cassandra.schema.Tables; -import org.apache.cassandra.utils.ByteBufferUtil; - -public class ReadRepairTest -{ - static Keyspace ks; - static ColumnFamilyStore cfs; - static TableMetadata cfm; - static InetAddressAndPort target1; - static InetAddressAndPort target2; - static InetAddressAndPort target3; - static List<InetAddressAndPort> targets; - - private static class InstrumentedReadRepairHandler extends BlockingPartitionRepair - { - public InstrumentedReadRepairHandler(Keyspace keyspace, DecoratedKey key, ConsistencyLevel consistency, Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor, InetAddressAndPort[] participants) - { - super(keyspace, key, consistency, repairs, maxBlockFor, participants); - } - - Map<InetAddressAndPort, Mutation> mutationsSent = new HashMap<>(); - - protected void sendRR(MessageOut<Mutation> message, InetAddressAndPort endpoint) - { - mutationsSent.put(endpoint, message.payload); - } - - List<InetAddressAndPort> candidates = targets; - - protected List<InetAddressAndPort> getCandidateEndpoints() - { - return candidates; - } - - @Override - protected boolean isLocal(InetAddressAndPort endpoint) - { - return targets.contains(endpoint); - } - } - - static long now = TimeUnit.NANOSECONDS.toMicros(System.nanoTime()); - static DecoratedKey key; - static Cell cell1; - static Cell cell2; - static Cell cell3; - static Mutation resolved; - - private static void assertRowsEqual(Row expected, Row actual) - { - try - { - Assert.assertEquals(expected == null, actual == null); - if (expected == null) - return; - Assert.assertEquals(expected.clustering(), actual.clustering()); - Assert.assertEquals(expected.deletion(), actual.deletion()); - Assert.assertArrayEquals(Iterables.toArray(expected.cells(), Cell.class), Iterables.toArray(expected.cells(), Cell.class)); - } catch (Throwable t) - { - throw new AssertionError(String.format("Row comparison failed, expected %s got %s", expected, actual), t); - } - } - - @BeforeClass - public static void setUpClass() throws Throwable - { - SchemaLoader.loadSchema(); - String ksName = "ks"; - - cfm = CreateTableStatement.parse("CREATE TABLE tbl (k int primary key, v text)", ksName).build(); - KeyspaceMetadata ksm = KeyspaceMetadata.create(ksName, KeyspaceParams.simple(3), Tables.of(cfm)); - MigrationManager.announceNewKeyspace(ksm, false); - - ks = Keyspace.open(ksName); - cfs = ks.getColumnFamilyStore("tbl"); - - cfs.sampleLatencyNanos = 0; - - target1 = InetAddressAndPort.getByName("127.0.0.255"); - target2 = InetAddressAndPort.getByName("127.0.0.254"); - target3 = InetAddressAndPort.getByName("127.0.0.253"); - - targets = ImmutableList.of(target1, target2, target3); - - // default test values - key = dk(5); - cell1 = cell("v", "val1", now); - cell2 = cell("v", "val2", now); - cell3 = cell("v", "val3", now); - resolved = mutation(cell1, cell2); - } - - private static DecoratedKey dk(int v) - { - return DatabaseDescriptor.getPartitioner().decorateKey(ByteBufferUtil.bytes(v)); - } - - private static Cell cell(String name, String value, long timestamp) - { - return BufferCell.live(cfm.getColumn(ColumnIdentifier.getInterned(name, false)), timestamp, ByteBufferUtil.bytes(value)); - } - - private static Mutation mutation(Cell... cells) - { - Row.Builder builder = BTreeRow.unsortedBuilder(); - builder.newRow(Clustering.EMPTY); - for (Cell cell: cells) - { - builder.addCell(cell); - } - return new Mutation(PartitionUpdate.singleRowUpdate(cfm, key, builder.build())); - } - - private static InstrumentedReadRepairHandler createRepairHandler(Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor, Collection<InetAddressAndPort> participants) - { - InetAddressAndPort[] participantArray = new InetAddressAndPort[participants.size()]; - participants.toArray(participantArray); - return new InstrumentedReadRepairHandler(ks, key, ConsistencyLevel.LOCAL_QUORUM, repairs, maxBlockFor, participantArray); - } - - private static InstrumentedReadRepairHandler createRepairHandler(Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor) - { - return createRepairHandler(repairs, maxBlockFor, repairs.keySet()); - } - - @Test - public void consistencyLevelTest() throws Exception - { - Assert.assertTrue(ConsistencyLevel.QUORUM.satisfies(ConsistencyLevel.QUORUM, ks)); - Assert.assertTrue(ConsistencyLevel.THREE.satisfies(ConsistencyLevel.QUORUM, ks)); - Assert.assertTrue(ConsistencyLevel.TWO.satisfies(ConsistencyLevel.QUORUM, ks)); - Assert.assertFalse(ConsistencyLevel.ONE.satisfies(ConsistencyLevel.QUORUM, ks)); - Assert.assertFalse(ConsistencyLevel.ANY.satisfies(ConsistencyLevel.QUORUM, ks)); - } - - private static void assertMutationEqual(Mutation expected, Mutation actual) - { - Assert.assertEquals(expected.getKeyspaceName(), actual.getKeyspaceName()); - Assert.assertEquals(expected.key(), actual.key()); - Assert.assertEquals(expected.key(), actual.key()); - PartitionUpdate expectedUpdate = Iterables.getOnlyElement(expected.getPartitionUpdates()); - PartitionUpdate actualUpdate = Iterables.getOnlyElement(actual.getPartitionUpdates()); - assertRowsEqual(Iterables.getOnlyElement(expectedUpdate), Iterables.getOnlyElement(actualUpdate)); - } - - @Test - public void additionalMutationRequired() throws Exception - { - - Mutation repair1 = mutation(cell2); - Mutation repair2 = mutation(cell1); - - // check that the correct repairs are calculated - Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); - repairs.put(target1, repair1); - repairs.put(target2, repair2); - - - InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2); - - Assert.assertTrue(handler.mutationsSent.isEmpty()); - - // check that the correct mutations are sent - handler.sendInitialRepairs(); - Assert.assertEquals(2, handler.mutationsSent.size()); - assertMutationEqual(repair1, handler.mutationsSent.get(target1)); - assertMutationEqual(repair2, handler.mutationsSent.get(target2)); - - // check that a combined mutation is speculatively sent to the 3rd target - handler.mutationsSent.clear(); - handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS); - Assert.assertEquals(1, handler.mutationsSent.size()); - assertMutationEqual(resolved, handler.mutationsSent.get(target3)); - - // check repairs stop blocking after receiving 2 acks - Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - handler.ack(target1); - Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - handler.ack(target3); - Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - - } - - /** - * If we've received enough acks, we shouldn't send any additional mutations - */ - @Test - public void noAdditionalMutationRequired() throws Exception - { - Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); - repairs.put(target1, mutation(cell2)); - repairs.put(target2, mutation(cell1)); - - InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2); - handler.sendInitialRepairs(); - handler.ack(target1); - handler.ack(target2); - - // both replicas have acked, we shouldn't send anything else out - handler.mutationsSent.clear(); - handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS); - Assert.assertTrue(handler.mutationsSent.isEmpty()); - } - - /** - * If there are no additional nodes we can send mutations to, we... shouldn't - */ - @Test - public void noAdditionalMutationPossible() throws Exception - { - Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); - repairs.put(target1, mutation(cell2)); - repairs.put(target2, mutation(cell1)); - - InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2); - handler.sendInitialRepairs(); - - // we've already sent mutations to all candidates, so we shouldn't send any more - handler.candidates = Lists.newArrayList(target1, target2); - handler.mutationsSent.clear(); - handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS); - Assert.assertTrue(handler.mutationsSent.isEmpty()); - } - - /** - * If we didn't send a repair to a replica because there wasn't a diff with the - * resolved column family, we shouldn't send it a speculative mutation - */ - @Test - public void mutationsArentSentToInSyncNodes() throws Exception - { - Mutation repair1 = mutation(cell2); - - Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); - repairs.put(target1, repair1); - Collection<InetAddressAndPort> participants = Lists.newArrayList(target1, target2); - - // check that the correct initial mutations are sent out - InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2, participants); - handler.sendInitialRepairs(); - Assert.assertEquals(1, handler.mutationsSent.size()); - Assert.assertTrue(handler.mutationsSent.containsKey(target1)); - - // check that speculative mutations aren't sent to target2 - handler.mutationsSent.clear(); - handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS); - Assert.assertEquals(1, handler.mutationsSent.size()); - Assert.assertTrue(handler.mutationsSent.containsKey(target3)); - } - - @Test - public void onlyBlockOnQuorum() - { - Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); - repairs.put(target1, mutation(cell1)); - repairs.put(target2, mutation(cell2)); - repairs.put(target3, mutation(cell3)); - Assert.assertEquals(3, repairs.size()); - - InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2); - handler.sendInitialRepairs(); - - Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - handler.ack(target1); - Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - - // here we should stop blocking, even though we've sent 3 repairs - handler.ack(target2); - Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - - } - - /** - * For dc local consistency levels, noop mutations and responses from remote dcs should not affect effective blockFor - */ - @Test - public void remoteDCTest() throws Exception - { - Map<InetAddressAndPort, Mutation> repairs = new HashMap<>(); - repairs.put(target1, mutation(cell1)); - - - InetAddressAndPort remote1 = InetAddressAndPort.getByName("10.0.0.1"); - InetAddressAndPort remote2 = InetAddressAndPort.getByName("10.0.0.2"); - repairs.put(remote1, mutation(cell1)); - - Collection<InetAddressAndPort> participants = Lists.newArrayList(target1, target2, remote1, remote2); - - InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2, participants); - handler.sendInitialRepairs(); - Assert.assertEquals(2, handler.mutationsSent.size()); - Assert.assertTrue(handler.mutationsSent.containsKey(target1)); - Assert.assertTrue(handler.mutationsSent.containsKey(remote1)); - - Assert.assertEquals(1, handler.waitingOn()); - Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - - handler.ack(remote1); - Assert.assertEquals(1, handler.waitingOn()); - Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - - handler.ack(target1); - Assert.assertEquals(0, handler.waitingOn()); - Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS)); - } -} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
