mirror of https://github.com/apache/cassandra
Make monotonic read / read repair configurable
Patch by Blake Eggleston; Reviewed by Aleksey Yeschenko for CASSANDRA-14635
This commit is contained in:
Blake Eggleston
parent
df892a38ff
commit
f25a765b6d
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
#####################
|
||||
|
||||
|
|
|
|||
|
|
@ -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>')]
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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"));
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
protected final Queue<BlockingPartitionRepair> repairs = new ConcurrentLinkedQueue<>();
|
||||
|
||||
private final Queue<BlockingPartitionRepair> repairs = new ConcurrentLinkedQueue<>();
|
||||
|
||||
private volatile DigestRepair digestRepair = null;
|
||||
|
||||
private static class DigestRepair
|
||||
public BlockingReadRepair(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency)
|
||||
{
|
||||
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 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()
|
||||
@Override
|
||||
Meter getRepairMeter()
|
||||
{
|
||||
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()
|
||||
{
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
{
|
||||
|
||||
}
|
||||
}
|
||||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
|
@ -20,50 +20,29 @@ 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.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.db.ReadCommand;
|
||||
import org.apache.cassandra.dht.Token;
|
||||
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;
|
||||
import org.apache.cassandra.service.reads.ReadCallback;
|
||||
|
||||
public class ReadRepairTest
|
||||
public class BlockingReadRepairTest extends AbstractReadRepairTest
|
||||
{
|
||||
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
|
||||
{
|
||||
|
|
@ -93,77 +72,10 @@ public class ReadRepairTest
|
|||
}
|
||||
}
|
||||
|
||||
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()));
|
||||
configureClass(ReadRepairStrategy.BLOCKING);
|
||||
}
|
||||
|
||||
private static InstrumentedReadRepairHandler createRepairHandler(Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor, Collection<InetAddressAndPort> participants)
|
||||
|
|
@ -178,6 +90,49 @@ public class ReadRepairTest
|
|||
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
|
||||
{
|
||||
|
|
@ -188,15 +143,6 @@ public class ReadRepairTest
|
|||
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
|
||||
|
|
@ -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();
|
||||
}
|
||||
|
|
@ -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[]{});
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue