natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Eliminate background repair and probablistic read_repair_chance table options 

patch by Aleksey Yeschenko; reviewed by Blake Eggleston for
CASSANDRA-13910
This commit is contained in:
Aleksey Yeshchenko 2018-03-28 12:47:09 +01:00
parent 88c54537d0
commit 2fcd29b830
36 changed files with 59 additions and 440 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -1,4 +1,6 @@
4.0
* Eliminate background repair and probablistic read_repair_chance table options
(CASSANDRA-13910)
* Bind to correct local address in 4.0 streaming (CASSANDRA-14362)
* Use standard Amazon naming for datacenter and rack in Ec2Snitch (CASSANDRA-7839)
* Fix junit failure for SSTableReaderTest (CASSANDRA-14387)

3
NEWS.txt
View File

@ -145,6 +145,9 @@ Upgrading
match the Amazon names. There is now a new option in conf/cassandra-rackdc.properties
that lets users enable the correct names for new clusters, or use the legacy
names for existing clusters. See conf/cassandra-rackdc.properties for details.
- Background repair has been removed. dclocal_read_repair_chance and
read_repair_chance table options have been removed and are now rejected.
See CASSANDRA-13910 for details.
Materialized Views
-------------------

2
conf/cassandra.yaml
View File

@ -917,7 +917,7 @@ dynamic_snitch_update_interval_in_ms: 100
# controls how often to reset all host scores, allowing a bad host to
# possibly recover
dynamic_snitch_reset_interval_in_ms: 600000
# if set greater than zero and read_repair_chance is < 1.0, this will allow
# if set greater than zero, this will allow
# 'pinning' of replicas to hosts in order to increase cache capacity.
# The badness threshold will control how much worse the pinned host has to be
# before the dynamic snitch will prefer other replicas over it. This is

8
doc/cql3/CQL.textile
View File

@ -250,8 +250,7 @@ CREATE TABLE monkeySpecies (
common_name text,
population varint,
average_size int
) WITH comment='Important biological records'
AND read_repair_chance = 1.0;
) WITH comment='Important biological records';
CREATE TABLE timeline (
userid uuid,
@ -334,8 +333,6 @@ Table creation supports the following other @<property>@:
|_. option |_. kind |_. default |_. description|
|@comment@ | _simple_ | none | A free-form, human-readable comment.|
|@read_repair_chance@ | _simple_ | 0.1 | The probability with which to query extra nodes (e.g. more nodes than required by the consistency level) for the purpose of read repairs.|
|@dclocal_read_repair_chance@ | _simple_ | 0 | The probability with which to query extra nodes (e.g. more nodes than required by the consistency level) belonging to the same data center than the read coordinator for the purpose of read repairs.|
|@gc_grace_seconds@ | _simple_ | 864000 | Time to wait before garbage collecting tombstones (deletion markers).|
|@bloom_filter_fp_chance@ | _simple_ | 0.00075 | The target probability of false positive of the sstable bloom filters. Said bloom filters will be sized to provide the provided probability (thus lowering this value impact the size of bloom filters in-memory and on-disk)|
|@default_time_to_live@ | _simple_ | 0 | The default expiration time ("TTL") in seconds for a table.|
@ -411,8 +408,7 @@ ALTER TABLE addamsFamily
ADD gravesite varchar;
ALTER TABLE addamsFamily
WITH comment = 'A most excellent and useful column family'
AND read_repair_chance = 0.2;
WITH comment = 'A most excellent and useful column family';
p.
The @ALTER@ statement is used to manipulate table definitions. It allows for adding new columns, dropping existing ones, or updating the table options. As with table creation, @ALTER COLUMNFAMILY@ is allowed as an alias for @ALTER TABLE@.

8
doc/source/architecture/dynamo.rst
View File

@ -117,12 +117,8 @@ Write operations are always sent to all replicas, regardless of consistency leve
controls how many responses the coordinator waits for before responding to the client.
For read operations, the coordinator generally only issues read commands to enough replicas to satisfy the consistency
level. There are a couple of exceptions to this:
- Speculative retry may issue a redundant read request to an extra replica if the other replicas have not responded
within a specified time window.
- Based on ``read_repair_chance`` and ``dclocal_read_repair_chance`` (part of a table's schema), read requests may be
randomly sent to all replicas in order to repair potentially inconsistent data.
level, with one exception. Speculative retry may issue a redundant read request to an extra replica if the other replicas
have not responded within a specified time window.
Picking Consistency Levels
~~~~~~~~~~~~~~~~~~~~~~~~~~

15
doc/source/cql/ddl.rst
View File

@ -186,8 +186,7 @@ For instance::
common_name text,
population varint,
average_size int
) WITH comment='Important biological records'
AND read_repair_chance = 1.0;
) WITH comment='Important biological records';
CREATE TABLE timeline (
userid uuid,
@ -453,15 +452,6 @@ A table supports the following options:
+================================+==========+=============+===========================================================+
| ``comment`` | *simple* | none | A free-form, human-readable comment. |
+--------------------------------+----------+-------------+-----------------------------------------------------------+
| ``read_repair_chance`` | *simple* | 0 | The probability with which to query extra nodes (e.g. |
| | | | more nodes than required by the consistency level) for |
| | | | the purpose of read repairs. |
+--------------------------------+----------+-------------+-----------------------------------------------------------+
| ``dclocal_read_repair_chance`` | *simple* | 0.1 | The probability with which to query extra nodes (e.g. |
| | | | more nodes than required by the consistency level) |
| | | | belonging to the same data center than the read |
| | | | coordinator for the purpose of read repairs. |
+--------------------------------+----------+-------------+-----------------------------------------------------------+
| ``speculative_retry`` | *simple* | 99PERCENTILE| :ref:`Speculative retry options |
| | | | <speculative-retry-options>`. |
+--------------------------------+----------+-------------+-----------------------------------------------------------+
@ -636,8 +626,7 @@ For instance::
ALTER TABLE addamsFamily ADD gravesite varchar;
ALTER TABLE addamsFamily
WITH comment = 'A most excellent and useful table'
AND read_repair_chance = 0.2;
WITH comment = 'A most excellent and useful table';
The ``ALTER TABLE`` statement can:

3
doc/source/operating/compaction.rst
View File

@ -432,8 +432,7 @@ order, with new data and old data in the same SSTable. Out of order data can app
While TWCS tries to minimize the impact of comingled data, users should attempt to avoid this behavior. Specifically,
users should avoid queries that explicitly set the timestamp via CQL ``USING TIMESTAMP``. Additionally, users should run
frequent repairs (which streams data in such a way that it does not become comingled), and disable background read
repair by setting the table's ``read_repair_chance`` and ``dclocal_read_repair_chance`` to 0.
frequent repairs (which streams data in such a way that it does not become comingled).
Changing TimeWindowCompactionStrategy Options
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

4
doc/source/operating/snitch.rst
View File

@ -35,8 +35,8 @@ configured with the following properties on ``cassandra.yaml``:
- ``dynamic_snitch``: whether the dynamic snitch should be enabled or disabled.
- ``dynamic_snitch_update_interval_in_ms``: controls how often to perform the more expensive part of host score
calculation.
- ``dynamic_snitch_reset_interval_in_ms``: if set greater than zero and read_repair_chance is < 1.0, this will allow
'pinning' of replicas to hosts in order to increase cache capacity.
- ``dynamic_snitch_reset_interval_in_ms``: if set greater than zero, this will allow 'pinning' of replicas to hosts
in order to increase cache capacity.
- ``dynamic_snitch_badness_threshold:``: The badness threshold will control how much worse the pinned host has to be
before the dynamic snitch will prefer other replicas over it. This is expressed as a double which represents a
percentage. Thus, a value of 0.2 means Cassandra would continue to prefer the static snitch values until the pinned

5
pylib/cqlshlib/cql3handling.py
View File

@ -44,11 +44,9 @@ class Cql3ParsingRuleSet(CqlParsingRuleSet):
columnfamily_layout_options = (
('bloom_filter_fp_chance', None),
('comment', None),
('dclocal_read_repair_chance', 'local_read_repair_chance'),
('gc_grace_seconds', None),
('min_index_interval', None),
('max_index_interval', None),
('read_repair_chance', None),
('default_time_to_live', None),
('speculative_retry', None),
('memtable_flush_period_in_ms', None),
@ -503,8 +501,7 @@ def cf_prop_val_completer(ctxt, cass):
return ["{'keys': '"]
if any(this_opt == opt[0] for opt in CqlRuleSet.obsolete_cf_options):
return ["'<obsolete_option>'"]
if this_opt in ('read_repair_chance', 'bloom_filter_fp_chance',
'dclocal_read_repair_chance'):
if this_opt == 'bloom_filter_fp_chance':
return [Hint('<float_between_0_and_1>')]
if this_opt in ('min_compaction_threshold', 'max_compaction_threshold',
'gc_grace_seconds', 'min_index_interval', 'max_index_interval'):

9
pylib/cqlshlib/test/test_cqlsh_completion.py
View File

@ -589,21 +589,19 @@ class TestCqlshCompletion(CqlshCompletionCase):
self.trycompletions(prefix + ' new_table (col_a int PRIMARY KEY) WITH ',
choices=['bloom_filter_fp_chance', 'compaction',
'compression',
'dclocal_read_repair_chance',
'default_time_to_live', 'gc_grace_seconds',
'max_index_interval',
'memtable_flush_period_in_ms',
'read_repair_chance', 'CLUSTERING',
'CLUSTERING',
'COMPACT', 'caching', 'comment',
'min_index_interval', 'speculative_retry', 'cdc'])
self.trycompletions(prefix + ' new_table (col_a int PRIMARY KEY) WITH ',
choices=['bloom_filter_fp_chance', 'compaction',
'compression',
'dclocal_read_repair_chance',
'default_time_to_live', 'gc_grace_seconds',
'max_index_interval',
'memtable_flush_period_in_ms',
'read_repair_chance', 'CLUSTERING',
'CLUSTERING',
'COMPACT', 'caching', 'comment',
'min_index_interval', 'speculative_retry', 'cdc'])
self.trycompletions(prefix + ' new_table (col_a int PRIMARY KEY) WITH bloom_filter_fp_chance ',
@ -647,11 +645,10 @@ class TestCqlshCompletion(CqlshCompletionCase):
+ "{'class': 'SizeTieredCompactionStrategy'} AND ",
choices=['bloom_filter_fp_chance', 'compaction',
'compression',
'dclocal_read_repair_chance',
'default_time_to_live', 'gc_grace_seconds',
'max_index_interval',
'memtable_flush_period_in_ms',
'read_repair_chance', 'CLUSTERING',
'CLUSTERING',
'COMPACT', 'caching', 'comment',
'min_index_interval', 'speculative_retry', 'cdc'])
self.trycompletions(prefix + " new_table (col_a int PRIMARY KEY) WITH compaction = "

2
pylib/cqlshlib/test/test_cqlsh_output.py
View File

@ -617,13 +617,11 @@ class TestCqlshOutput(BaseTestCase):
AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
AND crc_check_chance = 1.0
AND dclocal_read_repair_chance = 0.1
AND default_time_to_live = 0
AND gc_grace_seconds = 864000
AND max_index_interval = 2048
AND memtable_flush_period_in_ms = 0
AND min_index_interval = 128
AND read_repair_chance = 0.0
AND speculative_retry = '99PERCENTILE';
""" % quote_name(get_keyspace()))

1
src/java/org/apache/cassandra/auth/AuthKeyspace.java
View File

@ -84,7 +84,6 @@ public final class AuthKeyspace
return CreateTableStatement.parse(format(cql, name), SchemaConstants.AUTH_KEYSPACE_NAME)
.id(TableId.forSystemTable(SchemaConstants.AUTH_KEYSPACE_NAME, name))
.comment(description)
.dcLocalReadRepairChance(0.0)
.gcGraceSeconds((int) TimeUnit.DAYS.toSeconds(90))
.build();
}

6
src/java/org/apache/cassandra/cql3/statements/TableAttributes.java
View File

@ -104,9 +104,6 @@ public final class TableAttributes extends PropertyDefinitions
builder.compression(CompressionParams.fromMap(getMap(Option.COMPRESSION)));
}
if (hasOption(Option.DCLOCAL_READ_REPAIR_CHANCE))
builder.dcLocalReadRepairChance(getDouble(Option.DCLOCAL_READ_REPAIR_CHANCE));
if (hasOption(Option.DEFAULT_TIME_TO_LIVE))
builder.defaultTimeToLive(getInt(Option.DEFAULT_TIME_TO_LIVE));
@ -122,9 +119,6 @@ public final class TableAttributes extends PropertyDefinitions
if (hasOption(Option.MIN_INDEX_INTERVAL))
builder.minIndexInterval(getInt(Option.MIN_INDEX_INTERVAL));
if (hasOption(Option.READ_REPAIR_CHANCE))
builder.readRepairChance(getDouble(Option.READ_REPAIR_CHANCE));
if (hasOption(Option.SPECULATIVE_RETRY))
builder.speculativeRetry(SpeculativeRetryPolicy.fromString(getString(Option.SPECULATIVE_RETRY)));

47
src/java/org/apache/cassandra/db/ConsistencyLevel.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.db;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@ -30,7 +29,6 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.reads.ReadRepairDecision;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.UnavailableException;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
@ -176,11 +174,6 @@ public enum ConsistencyLevel
}
public List<InetAddressAndPort> filterForQuery(Keyspace keyspace, List<InetAddressAndPort> liveEndpoints)
{
return filterForQuery(keyspace, liveEndpoints, ReadRepairDecision.NONE);
}
public List<InetAddressAndPort> filterForQuery(Keyspace keyspace, List<InetAddressAndPort> liveEndpoints, ReadRepairDecision readRepair)
{
/*
* If we are doing an each quorum query, we have to make sure that the endpoints we select
@ -188,7 +181,7 @@ public enum ConsistencyLevel
* we should fall through and grab a quorum in the replication strategy.
*/
if (this == EACH_QUORUM && keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy)
return filterForEachQuorum(keyspace, liveEndpoints, readRepair);
return filterForEachQuorum(keyspace, liveEndpoints);
/*
* Endpoints are expected to be restricted to live replicas, sorted by snitch preference.
@ -197,42 +190,15 @@ public enum ConsistencyLevel
* the blockFor first ones).
*/
if (isDCLocal)
Collections.sort(liveEndpoints, DatabaseDescriptor.getLocalComparator());
liveEndpoints.sort(DatabaseDescriptor.getLocalComparator());
switch (readRepair)
{
case NONE:
return liveEndpoints.subList(0, Math.min(liveEndpoints.size(), blockFor(keyspace)));
case GLOBAL:
return liveEndpoints;
case DC_LOCAL:
List<InetAddressAndPort> local = new ArrayList<>();
List<InetAddressAndPort> other = new ArrayList<>();
for (InetAddressAndPort add : liveEndpoints)
{
if (isLocal(add))
local.add(add);
else
other.add(add);
}
// check if blockfor more than we have localep's
int blockFor = blockFor(keyspace);
if (local.size() < blockFor)
local.addAll(other.subList(0, Math.min(blockFor - local.size(), other.size())));
return local;
default:
throw new AssertionError();
}
return liveEndpoints.subList(0, Math.min(liveEndpoints.size(), blockFor(keyspace)));
}
private List<InetAddressAndPort> filterForEachQuorum(Keyspace keyspace, List<InetAddressAndPort> liveEndpoints, ReadRepairDecision readRepair)
private List<InetAddressAndPort> filterForEachQuorum(Keyspace keyspace, List<InetAddressAndPort> liveEndpoints)
{
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
// quickly drop out if read repair is GLOBAL, since we just use all of the live endpoints
if (readRepair == ReadRepairDecision.GLOBAL)
return liveEndpoints;
Map<String, List<InetAddressAndPort>> dcsEndpoints = new HashMap<>();
for (String dc: strategy.getDatacenters())
dcsEndpoints.put(dc, new ArrayList<>());
@ -247,10 +213,7 @@ public enum ConsistencyLevel
for (Map.Entry<String, List<InetAddressAndPort>> dcEndpoints : dcsEndpoints.entrySet())
{
List<InetAddressAndPort> dcEndpoint = dcEndpoints.getValue();
if (readRepair == ReadRepairDecision.DC_LOCAL && dcEndpoints.getKey().equals(DatabaseDescriptor.getLocalDataCenter()))
waitSet.addAll(dcEndpoint);
else
waitSet.addAll(dcEndpoint.subList(0, Math.min(localQuorumFor(keyspace, dcEndpoints.getKey()), dcEndpoint.size())));
waitSet.addAll(dcEndpoint.subList(0, Math.min(localQuorumFor(keyspace, dcEndpoints.getKey()), dcEndpoint.size())));
}
return waitSet;

1
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -370,7 +370,6 @@ public final class SystemKeyspace
{
return CreateTableStatement.parse(format(cql, table), SchemaConstants.SYSTEM_KEYSPACE_NAME)
.id(TableId.forSystemTable(SchemaConstants.SYSTEM_KEYSPACE_NAME, table))
.dcLocalReadRepairChance(0.0)
.gcGraceSeconds(0)
.memtableFlushPeriod((int) TimeUnit.HOURS.toMillis(1))
.comment(description);

2
src/java/org/apache/cassandra/db/TableCQLHelper.java
View File

@ -303,14 +303,12 @@ public class TableCQLHelper
StringBuilder builder = new StringBuilder();
builder.append("bloom_filter_fp_chance = ").append(tableParams.bloomFilterFpChance);
builder.append("\n\tAND dclocal_read_repair_chance = ").append(tableParams.dcLocalReadRepairChance);
builder.append("\n\tAND crc_check_chance = ").append(tableParams.crcCheckChance);
builder.append("\n\tAND default_time_to_live = ").append(tableParams.defaultTimeToLive);
builder.append("\n\tAND gc_grace_seconds = ").append(tableParams.gcGraceSeconds);
builder.append("\n\tAND min_index_interval = ").append(tableParams.minIndexInterval);
builder.append("\n\tAND max_index_interval = ").append(tableParams.maxIndexInterval);
builder.append("\n\tAND memtable_flush_period_in_ms = ").append(tableParams.memtableFlushPeriodInMs);
builder.append("\n\tAND read_repair_chance = ").append(tableParams.readRepairChance);
builder.append("\n\tAND speculative_retry = '").append(tableParams.speculativeRetry).append("'");
builder.append("\n\tAND comment = ").append(singleQuote(tableParams.comment));
builder.append("\n\tAND caching = ").append(toCQL(tableParams.caching.asMap()));

3
src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java
View File

@ -29,6 +29,9 @@ public class ReadRepairMetrics
private static final MetricNameFactory factory = new DefaultNameFactory("ReadRepair");
public static final Meter repairedBlocking = Metrics.meter(factory.createMetricName("RepairedBlocking"));
@Deprecated
public static final Meter repairedBackground = Metrics.meter(factory.createMetricName("RepairedBackground"));
@Deprecated
public static final Meter attempted = Metrics.meter(factory.createMetricName("Attempted"));
}

1
src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java
View File

@ -123,7 +123,6 @@ public final class SystemDistributedKeyspace
{
return CreateTableStatement.parse(format(cql, table), SchemaConstants.DISTRIBUTED_KEYSPACE_NAME)
.id(TableId.forSystemTable(SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, table))
.dcLocalReadRepairChance(0.0)
.comment(description)
.build();
}

15
src/java/org/apache/cassandra/schema/SchemaKeyspace.java
View File

@ -123,7 +123,7 @@ public final class SchemaKeyspace
+ "compaction frozen<map<text, text>>,"
+ "compression frozen<map<text, text>>,"
+ "crc_check_chance double,"
+ "dclocal_read_repair_chance double,"
+ "dclocal_read_repair_chance double," // no longer used, left for drivers' sake
+ "default_time_to_live int,"
+ "extensions frozen<map<text, blob>>,"
+ "flags frozen<set<text>>," // SUPER, COUNTER, DENSE, COMPOUND
@ -132,7 +132,7 @@ public final class SchemaKeyspace
+ "max_index_interval int,"
+ "memtable_flush_period_in_ms int,"
+ "min_index_interval int,"
+ "read_repair_chance double,"
+ "read_repair_chance double," // no longer used, left for drivers' sake
+ "speculative_retry text,"
+ "cdc boolean,"
+ "PRIMARY KEY ((keyspace_name), table_name))");
@ -188,7 +188,7 @@ public final class SchemaKeyspace
+ "compaction frozen<map<text, text>>,"
+ "compression frozen<map<text, text>>,"
+ "crc_check_chance double,"
+ "dclocal_read_repair_chance double,"
+ "dclocal_read_repair_chance double," // no longer used, left for drivers' sake
+ "default_time_to_live int,"
+ "extensions frozen<map<text, blob>>,"
+ "gc_grace_seconds int,"
@ -197,7 +197,7 @@ public final class SchemaKeyspace
+ "max_index_interval int,"
+ "memtable_flush_period_in_ms int,"
+ "min_index_interval int,"
+ "read_repair_chance double,"
+ "read_repair_chance double," // no longer used, left for drivers' sake
+ "speculative_retry text,"
+ "cdc boolean,"
+ "PRIMARY KEY ((keyspace_name), view_name))");
@ -258,7 +258,6 @@ public final class SchemaKeyspace
{
return CreateTableStatement.parse(format(cql, name), SchemaConstants.SCHEMA_KEYSPACE_NAME)
.id(TableId.forSystemTable(SchemaConstants.SCHEMA_KEYSPACE_NAME, name))
.dcLocalReadRepairChance(0.0)
.gcGraceSeconds((int) TimeUnit.DAYS.toSeconds(7))
.memtableFlushPeriod((int) TimeUnit.HOURS.toMillis(1))
.comment(description)
@ -524,13 +523,13 @@ public final class SchemaKeyspace
{
builder.add("bloom_filter_fp_chance", params.bloomFilterFpChance)
.add("comment", params.comment)
.add("dclocal_read_repair_chance", params.dcLocalReadRepairChance)
.add("dclocal_read_repair_chance", 0.0) // no longer used, left for drivers' sake
.add("default_time_to_live", params.defaultTimeToLive)
.add("gc_grace_seconds", params.gcGraceSeconds)
.add("max_index_interval", params.maxIndexInterval)
.add("memtable_flush_period_in_ms", params.memtableFlushPeriodInMs)
.add("min_index_interval", params.minIndexInterval)
.add("read_repair_chance", params.readRepairChance)
.add("read_repair_chance", 0.0) // no longer used, left for drivers' sake
.add("speculative_retry", params.speculativeRetry.toString())
.add("crc_check_chance", params.crcCheckChance)
.add("caching", params.caching.asMap())
@ -994,14 +993,12 @@ public final class SchemaKeyspace
.comment(row.getString("comment"))
.compaction(CompactionParams.fromMap(row.getFrozenTextMap("compaction")))
.compression(CompressionParams.fromMap(row.getFrozenTextMap("compression")))
.dcLocalReadRepairChance(row.getDouble("dclocal_read_repair_chance"))
.defaultTimeToLive(row.getInt("default_time_to_live"))
.extensions(row.getFrozenMap("extensions", UTF8Type.instance, BytesType.instance))
.gcGraceSeconds(row.getInt("gc_grace_seconds"))
.maxIndexInterval(row.getInt("max_index_interval"))
.memtableFlushPeriodInMs(row.getInt("memtable_flush_period_in_ms"))
.minIndexInterval(row.getInt("min_index_interval"))
.readRepairChance(row.getDouble("read_repair_chance"))
.crcCheckChance(row.getDouble("crc_check_chance"))
.speculativeRetry(SpeculativeRetryPolicy.fromString(row.getString("speculative_retry")))
.cdc(row.has("cdc") && row.getBoolean("cdc"))

18
src/java/org/apache/cassandra/schema/TableMetadata.java
View File

@ -523,11 +523,7 @@ public final class TableMetadata
public TableMetadata updateIndexTableMetadata(TableParams baseTableParams)
{
TableParams.Builder builder =
baseTableParams.unbuild()
.readRepairChance(0.0)
.dcLocalReadRepairChance(0.0)
.gcGraceSeconds(0);
TableParams.Builder builder = baseTableParams.unbuild().gcGraceSeconds(0);
// Depends on parent's cache setting, turn on its index table's cache.
// Row caching is never enabled; see CASSANDRA-5732
@ -689,12 +685,6 @@ public final class TableMetadata
return this;
}
public Builder dcLocalReadRepairChance(double val)
{
params.dcLocalReadRepairChance(val);
return this;
}
public Builder defaultTimeToLive(int val)
{
params.defaultTimeToLive(val);
@ -725,12 +715,6 @@ public final class TableMetadata
return this;
}
public Builder readRepairChance(double val)
{
params.readRepairChance(val);
return this;
}
public Builder crcCheckChance(double val)
{
params.crcCheckChance(val);

42
src/java/org/apache/cassandra/schema/TableParams.java
View File

@ -43,14 +43,12 @@ public final class TableParams
COMMENT,
COMPACTION,
COMPRESSION,
DCLOCAL_READ_REPAIR_CHANCE,
DEFAULT_TIME_TO_LIVE,
EXTENSIONS,
GC_GRACE_SECONDS,
MAX_INDEX_INTERVAL,
MEMTABLE_FLUSH_PERIOD_IN_MS,
MIN_INDEX_INTERVAL,
READ_REPAIR_CHANCE,
SPECULATIVE_RETRY,
CRC_CHECK_CHANCE,
CDC;
@ -63,8 +61,6 @@ public final class TableParams
}
public final String comment;
public final double readRepairChance;
public final double dcLocalReadRepairChance;
public final double bloomFilterFpChance;
public final double crcCheckChance;
public final int gcGraceSeconds;
@ -82,8 +78,6 @@ public final class TableParams
private TableParams(Builder builder)
{
comment = builder.comment;
readRepairChance = builder.readRepairChance;
dcLocalReadRepairChance = builder.dcLocalReadRepairChance;
bloomFilterFpChance = builder.bloomFilterFpChance == null
? builder.compaction.defaultBloomFilterFbChance()
: builder.bloomFilterFpChance;
@ -113,14 +107,12 @@ public final class TableParams
.comment(params.comment)
.compaction(params.compaction)
.compression(params.compression)
.dcLocalReadRepairChance(params.dcLocalReadRepairChance)
.crcCheckChance(params.crcCheckChance)
.defaultTimeToLive(params.defaultTimeToLive)
.gcGraceSeconds(params.gcGraceSeconds)
.maxIndexInterval(params.maxIndexInterval)
.memtableFlushPeriodInMs(params.memtableFlushPeriodInMs)
.minIndexInterval(params.minIndexInterval)
.readRepairChance(params.readRepairChance)
.speculativeRetry(params.speculativeRetry)
.extensions(params.extensions)
.cdc(params.cdc);
@ -145,20 +137,6 @@ public final class TableParams
bloomFilterFpChance);
}
if (dcLocalReadRepairChance < 0 || dcLocalReadRepairChance > 1.0)
{
fail("%s must be larger than or equal to 0 and smaller than or equal to 1.0 (got %s)",
Option.DCLOCAL_READ_REPAIR_CHANCE,
dcLocalReadRepairChance);
}
if (readRepairChance < 0 || readRepairChance > 1.0)
{
fail("%s must be larger than or equal to 0 and smaller than or equal to 1.0 (got %s)",
Option.READ_REPAIR_CHANCE,
readRepairChance);
}
if (crcCheckChance < 0 || crcCheckChance > 1.0)
{
fail("%s must be larger than or equal to 0 and smaller than or equal to 1.0 (got %s)",
@ -208,8 +186,6 @@ public final class TableParams
TableParams p = (TableParams) o;
return comment.equals(p.comment)
&& readRepairChance == p.readRepairChance
&& dcLocalReadRepairChance == p.dcLocalReadRepairChance
&& bloomFilterFpChance == p.bloomFilterFpChance
&& crcCheckChance == p.crcCheckChance
&& gcGraceSeconds == p.gcGraceSeconds
@ -229,8 +205,6 @@ public final class TableParams
public int hashCode()
{
return Objects.hashCode(comment,
readRepairChance,
dcLocalReadRepairChance,
bloomFilterFpChance,
crcCheckChance,
gcGraceSeconds,
@ -251,8 +225,6 @@ public final class TableParams
{
return MoreObjects.toStringHelper(this)
.add(Option.COMMENT.toString(), comment)
.add(Option.READ_REPAIR_CHANCE.toString(), readRepairChance)
.add(Option.DCLOCAL_READ_REPAIR_CHANCE.toString(), dcLocalReadRepairChance)
.add(Option.BLOOM_FILTER_FP_CHANCE.toString(), bloomFilterFpChance)
.add(Option.CRC_CHECK_CHANCE.toString(), crcCheckChance)
.add(Option.GC_GRACE_SECONDS.toString(), gcGraceSeconds)
@ -272,8 +244,6 @@ public final class TableParams
public static final class Builder
{
private String comment = "";
private double readRepairChance = 0.0;
private double dcLocalReadRepairChance = 0.1;
private Double bloomFilterFpChance;
public Double crcCheckChance = 1.0;
private int gcGraceSeconds = 864000; // 10 days
@ -303,18 +273,6 @@ public final class TableParams
return this;
}
public Builder readRepairChance(double val)
{
readRepairChance = val;
return this;
}
public Builder dcLocalReadRepairChance(double val)
{
dcLocalReadRepairChance = val;
return this;
}
public Builder bloomFilterFpChance(double val)
{
bloomFilterFpChance = val;

57
src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.service.reads;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Preconditions;
@ -39,12 +38,10 @@ import org.apache.cassandra.exceptions.ReadFailureException;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.UnavailableException;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.ReadRepairMetrics;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.reads.repair.ReadRepair;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageProxy.LocalReadRunnable;
import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.tracing.Tracing;
@ -162,22 +159,6 @@ public abstract class AbstractReadExecutor
*/
public abstract void executeAsync();
private static ReadRepairDecision newReadRepairDecision(TableMetadata metadata)
{
if (metadata.params.readRepairChance > 0d ||
metadata.params.dcLocalReadRepairChance > 0)
{
double chance = ThreadLocalRandom.current().nextDouble();
if (metadata.params.readRepairChance > chance)
return ReadRepairDecision.GLOBAL;
if (metadata.params.dcLocalReadRepairChance > chance)
return ReadRepairDecision.DC_LOCAL;
}
return ReadRepairDecision.NONE;
}
/**
* @return an executor appropriate for the configured speculative read policy
*/
@ -185,28 +166,17 @@ public abstract class AbstractReadExecutor
{
Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
List<InetAddressAndPort> allReplicas = StorageProxy.getLiveSortedEndpoints(keyspace, command.partitionKey());
// 11980: Excluding EACH_QUORUM reads from potential RR, so that we do not miscount DC responses
ReadRepairDecision repairDecision = consistencyLevel == ConsistencyLevel.EACH_QUORUM
? ReadRepairDecision.NONE
: newReadRepairDecision(command.metadata());
List<InetAddressAndPort> targetReplicas = consistencyLevel.filterForQuery(keyspace, allReplicas, repairDecision);
List<InetAddressAndPort> targetReplicas = consistencyLevel.filterForQuery(keyspace, allReplicas);
// Throw UAE early if we don't have enough replicas.
consistencyLevel.assureSufficientLiveNodes(keyspace, targetReplicas);
if (repairDecision != ReadRepairDecision.NONE)
{
Tracing.trace("Read-repair {}", repairDecision);
ReadRepairMetrics.attempted.mark();
}
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(command.metadata().id);
SpeculativeRetryPolicy retry = cfs.metadata().params.speculativeRetry;
// Speculative retry is disabled *OR*
// 11980: Disable speculative retry if using EACH_QUORUM in order to prevent miscounting DC responses
if (retry.equals(NeverSpeculativeRetryPolicy.INSTANCE)
| consistencyLevel == ConsistencyLevel.EACH_QUORUM)
if (retry.equals(NeverSpeculativeRetryPolicy.INSTANCE) || consistencyLevel == ConsistencyLevel.EACH_QUORUM)
return new NeverSpeculatingReadExecutor(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime, false);
// There are simply no extra replicas to speculate.
@ -216,28 +186,13 @@ public abstract class AbstractReadExecutor
if (targetReplicas.size() == allReplicas.size())
{
// CL.ALL, RRD.GLOBAL or RRD.DC_LOCAL and a single-DC.
// CL.ALL
// We are going to contact every node anyway, so ask for 2 full data requests instead of 1, for redundancy
// (same amount of requests in total, but we turn 1 digest request into a full blown data request).
return new AlwaysSpeculatingReadExecutor(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime);
}
// RRD.NONE or RRD.DC_LOCAL w/ multiple DCs.
InetAddressAndPort extraReplica = allReplicas.get(targetReplicas.size());
// With repair decision DC_LOCAL all replicas/target replicas may be in different order, so
// we might have to find a replacement that's not already in targetReplicas.
if (repairDecision == ReadRepairDecision.DC_LOCAL && targetReplicas.contains(extraReplica))
{
for (InetAddressAndPort address : allReplicas)
{
if (!targetReplicas.contains(address))
{
extraReplica = address;
break;
}
}
}
targetReplicas.add(extraReplica);
targetReplicas.add(allReplicas.get(targetReplicas.size()));
if (retry.equals(AlwaysSpeculativeRetryPolicy.INSTANCE))
return new AlwaysSpeculatingReadExecutor(keyspace, cfs, command, consistencyLevel, targetReplicas, queryStartNanoTime);
@ -445,7 +400,7 @@ public abstract class AbstractReadExecutor
else
{
Tracing.trace("Digest mismatch: Mismatch for key {}", getKey());
readRepair.startForegroundRepair(digestResolver, handler.endpoints, getContactedReplicas(), this::setResult);
readRepair.startRepair(digestResolver, handler.endpoints, getContactedReplicas(), this::setResult);
}
}
@ -453,7 +408,7 @@ public abstract class AbstractReadExecutor
{
try
{
readRepair.awaitForegroundRepairFinish();
readRepair.awaitRepair();
}
catch (ReadTimeoutException e)
{

61
src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java
View File

@ -1,61 +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;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.db.ReadResponse;
import org.apache.cassandra.net.IAsyncCallback;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.service.reads.DataResolver;
import org.apache.cassandra.utils.WrappedRunnable;
public class AsyncRepairCallback implements IAsyncCallback<ReadResponse>
{
private final DataResolver repairResolver;
private final int blockfor;
protected final AtomicInteger received = new AtomicInteger(0);
public AsyncRepairCallback(DataResolver repairResolver, int blockfor)
{
this.repairResolver = repairResolver;
this.blockfor = blockfor;
}
public void response(MessageIn<ReadResponse> message)
{
repairResolver.preprocess(message);
if (received.incrementAndGet() == blockfor)
{
StageManager.getStage(Stage.READ_REPAIR).execute(new WrappedRunnable()
{
protected void runMayThrow()
{
repairResolver.evaluateAllResponses();
}
});
}
}
public boolean isLatencyForSnitch()
{
return true;
}
}

29
src/java/org/apache/cassandra/service/reads/DataResolver.java
View File

@ -22,19 +22,18 @@ import java.util.*;
import com.google.common.base.Joiner;
import com.google.common.collect.Iterables;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.rows.RangeTombstoneMarker;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.db.transform.*;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.*;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.reads.repair.ReadRepair;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.transform.*;
import org.apache.cassandra.net.*;
import org.apache.cassandra.tracing.TraceState;
public class DataResolver extends ResponseResolver
{
@ -88,7 +87,7 @@ public class DataResolver extends ResponseResolver
*/
DataLimits.Counter mergedResultCounter =
command.limits().newCounter(command.nowInSec(), true, command.selectsFullPartition(), enforceStrictLiveness);
command.limits().newCounter(command.nowInSec(), true, command.selectsFullPartition(), enforceStrictLiveness);
UnfilteredPartitionIterator merged = mergeWithShortReadProtection(iters, sources, mergedResultCounter);
FilteredPartitions filtered = FilteredPartitions.filter(merged, new Filter(command.nowInSec(), command.metadata().enforceStrictLiveness()));
@ -110,27 +109,11 @@ public class DataResolver extends ResponseResolver
*/
if (!command.limits().isUnlimited())
for (int i = 0; i < results.size(); i++)
{
results.set(i, ShortReadProtection.extend(sources[i], results.get(i), command, mergedResultCounter, queryStartNanoTime, enforceStrictLiveness));
}
return UnfilteredPartitionIterators.merge(results, command.nowInSec(), wrapMergeListener(readRepair.getMergeListener(sources), sources));
}
public void evaluateAllResponses()
{
// We need to fully consume the results to trigger read repairs if appropriate
try (PartitionIterator iterator = resolve())
{
PartitionIterators.consume(iterator);
}
}
public void evaluateAllResponses(TraceState traceState)
{
evaluateAllResponses();
}
private String makeResponsesDebugString(DecoratedKey partitionKey)
{
return Joiner.on(",\n").join(Iterables.transform(getMessages(), m -> m.from + " => " + m.payload.toDebugString(command, partitionKey)));

8
src/java/org/apache/cassandra/service/reads/DigestResolver.java
View File

@ -78,14 +78,6 @@ public class DigestResolver extends ResponseResolver
return true;
}
public void evaluateAllResponses(TraceState traceState)
{
if (!responsesMatch())
{
readRepair.backgroundDigestRepair(traceState);
}
}
public boolean isDataPresent()
{
return dataResponse != null;

9
src/java/org/apache/cassandra/service/reads/ReadCallback.java
View File

@ -148,16 +148,9 @@ public class ReadCallback implements IAsyncCallbackWithFailure<ReadResponse>
int n = waitingFor(message.from)
? recievedUpdater.incrementAndGet(this)
: received;
if (n >= blockfor && resolver.isDataPresent())
{
condition.signalAll();
// kick off a background digest comparison if this is a result that (may have) arrived after
// the original resolve that get() kicks off as soon as the condition is signaled
if (blockfor < endpoints.size() && n == endpoints.size())
{
readRepair.maybeStartBackgroundRepair(resolver);
}
}
}
/**

23
src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java
View File

@ -1,23 +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;
public enum ReadRepairDecision
{
NONE, GLOBAL, DC_LOCAL;
}

6
src/java/org/apache/cassandra/service/reads/ResponseResolver.java
View File

@ -23,7 +23,6 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.*;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.service.reads.repair.ReadRepair;
import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.utils.concurrent.Accumulator;
public abstract class ResponseResolver
@ -47,11 +46,6 @@ public abstract class ResponseResolver
this.responses = new Accumulator<>(maxResponseCount);
}
/**
* Consume the accumulated responses, starting a read repair if neccesary
*/
public abstract void evaluateAllResponses(TraceState traceState);
public abstract boolean isDataPresent();
public void preprocess(MessageIn<ReadResponse> message)

34
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java
View File

@ -36,8 +36,6 @@ import com.google.common.util.concurrent.Futures;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ConsistencyLevel;
@ -53,12 +51,9 @@ import org.apache.cassandra.metrics.ReadRepairMetrics;
import org.apache.cassandra.net.AsyncOneResponse;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.reads.AsyncRepairCallback;
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.service.reads.ResponseResolver;
import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.tracing.Tracing;
/**
@ -225,7 +220,7 @@ public class BlockingReadRepair implements ReadRepair, RepairListener
return repair;
}
public void startForegroundRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
{
ReadRepairMetrics.repairedBlocking.mark();
@ -244,7 +239,7 @@ public class BlockingReadRepair implements ReadRepair, RepairListener
}
}
public void awaitForegroundRepairFinish() throws ReadTimeoutException
public void awaitRepair() throws ReadTimeoutException
{
if (digestRepair != null)
{
@ -252,29 +247,4 @@ public class BlockingReadRepair implements ReadRepair, RepairListener
digestRepair.resultConsumer.accept(digestRepair.dataResolver.resolve());
}
}
public void maybeStartBackgroundRepair(ResponseResolver resolver)
{
TraceState traceState = Tracing.instance.get();
if (traceState != null)
traceState.trace("Initiating read-repair");
StageManager.getStage(Stage.READ_REPAIR).execute(() -> resolver.evaluateAllResponses(traceState));
}
public void backgroundDigestRepair(TraceState traceState)
{
if (traceState != null)
traceState.trace("Digest mismatch");
if (logger.isDebugEnabled())
logger.debug("Digest mismatch");
ReadRepairMetrics.repairedBackground.mark();
Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
final DataResolver repairResolver = new DataResolver(keyspace, command, consistency, endpoints.size(), queryStartNanoTime, this);
AsyncRepairCallback repairHandler = new AsyncRepairCallback(repairResolver, endpoints.size());
for (InetAddressAndPort endpoint : endpoints)
MessagingService.instance().sendRR(command.createMessage(), endpoint, repairHandler);
}
}

17
src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java
View File

@ -26,8 +26,6 @@ import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.reads.DigestResolver;
import org.apache.cassandra.service.reads.ResponseResolver;
import org.apache.cassandra.tracing.TraceState;
public class NoopReadRepair implements ReadRepair
{
@ -40,23 +38,12 @@ public class NoopReadRepair implements ReadRepair
return UnfilteredPartitionIterators.MergeListener.NOOP;
}
public void startForegroundRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
{
resultConsumer.accept(digestResolver.getData());
}
public void awaitForegroundRepairFinish() throws ReadTimeoutException
public void awaitRepair() throws ReadTimeoutException
{
}
public void maybeStartBackgroundRepair(ResponseResolver resolver)
{
}
public void backgroundDigestRepair(TraceState traceState)
{
}
}

29
src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java
View File

@ -15,7 +15,6 @@
* 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;
@ -28,12 +27,9 @@ import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.reads.DigestResolver;
import org.apache.cassandra.service.reads.ResponseResolver;
import org.apache.cassandra.tracing.TraceState;
public interface ReadRepair
{
/**
* Used by DataResolver to generate corrections as the partition iterator is consumed
*/
@ -43,28 +39,15 @@ public interface ReadRepair
* Called when the digests from the initial read don't match. Reads may block on the
* repair started by this method.
*/
public void startForegroundRepair(DigestResolver digestResolver,
List<InetAddressAndPort> allEndpoints,
List<InetAddressAndPort> contactedEndpoints,
Consumer<PartitionIterator> resultConsumer);
public void startRepair(DigestResolver digestResolver,
List<InetAddressAndPort> allEndpoints,
List<InetAddressAndPort> contactedEndpoints,
Consumer<PartitionIterator> resultConsumer);
/**
* Wait for any operations started by {@link ReadRepair#startForegroundRepair} to complete
* @throws ReadTimeoutException
* Wait for any operations started by {@link ReadRepair#startRepair} to complete
*/
public void awaitForegroundRepairFinish() throws ReadTimeoutException;
/**
* Called when responses from all replicas have been received. Read will not block on this.
* @param resolver
*/
public void maybeStartBackgroundRepair(ResponseResolver resolver);
/**
* If {@link ReadRepair#maybeStartBackgroundRepair} was called with a {@link DigestResolver}, this will
* be called to perform a repair if there was a digest mismatch
*/
public void backgroundDigestRepair(TraceState traceState);
public void awaitRepair() throws ReadTimeoutException;
static ReadRepair create(ReadCommand command, List<InetAddressAndPort> endpoints, long queryStartNanoTime, ConsistencyLevel consistency)
{

1
src/java/org/apache/cassandra/tracing/TraceKeyspace.java
View File

@ -78,7 +78,6 @@ public final class TraceKeyspace
{
return CreateTableStatement.parse(format(cql, table), SchemaConstants.TRACE_KEYSPACE_NAME)
.id(TableId.forSystemTable(SchemaConstants.TRACE_KEYSPACE_NAME, table))
.dcLocalReadRepairChance(0.0)
.gcGraceSeconds(0)
.memtableFlushPeriod((int) TimeUnit.HOURS.toMillis(1))
.comment(description)

4
test/unit/org/apache/cassandra/cql3/validation/miscellaneous/OverflowTest.java
View File

@ -107,8 +107,6 @@ public class OverflowTest extends CQLTester
{
createTable("CREATE TABLE %s ( k int PRIMARY KEY, c int ) WITH "
+ "comment = 'My comment' "
+ "AND read_repair_chance = 0.5 "
+ "AND dclocal_read_repair_chance = 0.5 "
+ "AND gc_grace_seconds = 4 "
+ "AND bloom_filter_fp_chance = 0.01 "
+ "AND compaction = { 'class' : 'LeveledCompactionStrategy', 'sstable_size_in_mb' : 10, 'fanout_size' : 5 } "
@ -117,8 +115,6 @@ public class OverflowTest extends CQLTester
execute("ALTER TABLE %s WITH "
+ "comment = 'other comment' "
+ "AND read_repair_chance = 0.3 "
+ "AND dclocal_read_repair_chance = 0.3 "
+ "AND gc_grace_seconds = 100 "
+ "AND bloom_filter_fp_chance = 0.1 "
+ "AND compaction = { 'class': 'SizeTieredCompactionStrategy', 'min_sstable_size' : 42 } "

4
test/unit/org/apache/cassandra/db/TableCQLHelperTest.java
View File

@ -248,14 +248,12 @@ public class TableCQLHelperTest extends CQLTester
.comment("comment")
.compaction(CompactionParams.lcs(Collections.singletonMap("sstable_size_in_mb", "1")))
.compression(CompressionParams.lz4(1 << 16, 1 << 15))
.dcLocalReadRepairChance(0.2)
.crcCheckChance(0.3)
.defaultTimeToLive(4)
.gcGraceSeconds(5)
.minIndexInterval(6)
.maxIndexInterval(7)
.memtableFlushPeriod(8)
.readRepairChance(0.9)
.speculativeRetry(AlwaysSpeculativeRetryPolicy.INSTANCE)
.extensions(ImmutableMap.of("ext1", ByteBuffer.wrap("val1".getBytes())))
.recordColumnDrop(ColumnMetadata.regularColumn(keyspace, table, "reg1", AsciiType.instance),
@ -267,14 +265,12 @@ public class TableCQLHelperTest extends CQLTester
assertTrue(TableCQLHelper.getTableMetadataAsCQL(cfs.metadata(), true).endsWith(
"AND bloom_filter_fp_chance = 1.0\n" +
"\tAND dclocal_read_repair_chance = 0.2\n" +
"\tAND crc_check_chance = 0.3\n" +
"\tAND default_time_to_live = 4\n" +
"\tAND gc_grace_seconds = 5\n" +
"\tAND min_index_interval = 6\n" +
"\tAND max_index_interval = 7\n" +
"\tAND memtable_flush_period_in_ms = 8\n" +
"\tAND read_repair_chance = 0.9\n" +
"\tAND speculative_retry = 'ALWAYS'\n" +
"\tAND comment = 'comment'\n" +
"\tAND caching = { 'keys': 'ALL', 'rows_per_partition': 'NONE' }\n" +

2
test/unit/org/apache/cassandra/schema/MigrationManagerTest.java
View File

@ -100,7 +100,6 @@ public class MigrationManagerTest
.addPartitionKeyColumn("keys", BytesType.instance)
.addClusteringColumn("col", BytesType.instance)
.comment("No comment")
.readRepairChance(0.5)
.gcGraceSeconds(100000)
.compaction(CompactionParams.scts(ImmutableMap.of("min_threshold", "500", "max_threshold", "500")));
@ -573,7 +572,6 @@ public class MigrationManagerTest
.addClusteringColumn("col", UTF8Type.instance)
.addRegularColumn("val", UTF8Type.instance)
.comment(comment)
.readRepairChance(0.0)
.build();
}
}

18
test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java
View File

@ -30,8 +30,6 @@ import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.reads.DigestResolver;
import org.apache.cassandra.service.reads.ResponseResolver;
import org.apache.cassandra.tracing.TraceState;
public class TestableReadRepair implements ReadRepair, RepairListener
{
@ -66,25 +64,13 @@ public class TestableReadRepair implements ReadRepair, RepairListener
}
@Override
public void startForegroundRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
{
}
@Override
public void awaitForegroundRepairFinish() throws ReadTimeoutException
{
}
@Override
public void maybeStartBackgroundRepair(ResponseResolver resolver)
{
}
@Override
public void backgroundDigestRepair(TraceState traceState)
public void awaitRepair() throws ReadTimeoutException
{
}