Eliminate background repair and probablistic read_repair_chance table options
patch by Aleksey Yeschenko; reviewed by Blake Eggleston for CASSANDRA-13910 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/2fcd29b8 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/2fcd29b8 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/2fcd29b8 Branch: refs/heads/trunk Commit: 2fcd29b830e7b201e7047d283de385d5f1c427b5 Parents: 88c5453 Author: Aleksey Yeshchenko <[email protected]> Authored: Wed Mar 28 12:47:09 2018 +0100 Committer: Aleksey Yeshchenko <[email protected]> Committed: Wed Apr 18 11:41:17 2018 +0100 ---------------------------------------------------------------------- CHANGES.txt | 2 + NEWS.txt | 3 + conf/cassandra.yaml | 2 +- doc/cql3/CQL.textile | 8 +-- doc/source/architecture/dynamo.rst | 8 +-- doc/source/cql/ddl.rst | 15 +---- doc/source/operating/compaction.rst | 3 +- doc/source/operating/snitch.rst | 4 +- pylib/cqlshlib/cql3handling.py | 5 +- pylib/cqlshlib/test/test_cqlsh_completion.py | 9 +-- pylib/cqlshlib/test/test_cqlsh_output.py | 2 - .../org/apache/cassandra/auth/AuthKeyspace.java | 1 - .../cql3/statements/TableAttributes.java | 6 -- .../apache/cassandra/db/ConsistencyLevel.java | 47 ++------------- .../org/apache/cassandra/db/SystemKeyspace.java | 1 - .../org/apache/cassandra/db/TableCQLHelper.java | 2 - .../cassandra/metrics/ReadRepairMetrics.java | 3 + .../repair/SystemDistributedKeyspace.java | 1 - .../apache/cassandra/schema/SchemaKeyspace.java | 15 ++--- .../apache/cassandra/schema/TableMetadata.java | 18 +----- .../apache/cassandra/schema/TableParams.java | 42 -------------- .../service/reads/AbstractReadExecutor.java | 57 ++---------------- .../service/reads/AsyncRepairCallback.java | 61 -------------------- .../cassandra/service/reads/DataResolver.java | 29 ++-------- .../cassandra/service/reads/DigestResolver.java | 8 --- .../cassandra/service/reads/ReadCallback.java | 9 +-- .../service/reads/ReadRepairDecision.java | 23 -------- .../service/reads/ResponseResolver.java | 6 -- .../reads/repair/BlockingReadRepair.java | 34 +---------- .../service/reads/repair/NoopReadRepair.java | 17 +----- .../service/reads/repair/ReadRepair.java | 29 ++-------- .../apache/cassandra/tracing/TraceKeyspace.java | 1 - .../validation/miscellaneous/OverflowTest.java | 4 -- .../apache/cassandra/db/TableCQLHelperTest.java | 4 -- .../cassandra/schema/MigrationManagerTest.java | 2 - .../reads/repair/TestableReadRepair.java | 18 +----- 36 files changed, 59 insertions(+), 440 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index f2c9ffc..76321dd 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/NEWS.txt ---------------------------------------------------------------------- diff --git a/NEWS.txt b/NEWS.txt index 3be52dc..e8e88a4 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -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 ------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/conf/cassandra.yaml ---------------------------------------------------------------------- diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 1be6feb..d466072 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -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 http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/doc/cql3/CQL.textile ---------------------------------------------------------------------- diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile index d87e72a..8d55baa 100644 --- a/doc/cql3/CQL.textile +++ b/doc/cql3/CQL.textile @@ -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@. http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/doc/source/architecture/dynamo.rst ---------------------------------------------------------------------- diff --git a/doc/source/architecture/dynamo.rst b/doc/source/architecture/dynamo.rst index a7dbb87..365a695 100644 --- a/doc/source/architecture/dynamo.rst +++ b/doc/source/architecture/dynamo.rst @@ -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 ~~~~~~~~~~~~~~~~~~~~~~~~~~ http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/doc/source/cql/ddl.rst ---------------------------------------------------------------------- diff --git a/doc/source/cql/ddl.rst b/doc/source/cql/ddl.rst index be5e2c6..c8cedcf 100644 --- a/doc/source/cql/ddl.rst +++ b/doc/source/cql/ddl.rst @@ -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: http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/doc/source/operating/compaction.rst ---------------------------------------------------------------------- diff --git a/doc/source/operating/compaction.rst b/doc/source/operating/compaction.rst index cb6be45..ace9aa9 100644 --- a/doc/source/operating/compaction.rst +++ b/doc/source/operating/compaction.rst @@ -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 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/doc/source/operating/snitch.rst ---------------------------------------------------------------------- diff --git a/doc/source/operating/snitch.rst b/doc/source/operating/snitch.rst index 91bab29..5f6760a 100644 --- a/doc/source/operating/snitch.rst +++ b/doc/source/operating/snitch.rst @@ -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 http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/pylib/cqlshlib/cql3handling.py ---------------------------------------------------------------------- diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py index ca14a36..314c431 100644 --- a/pylib/cqlshlib/cql3handling.py +++ b/pylib/cqlshlib/cql3handling.py @@ -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'): http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/pylib/cqlshlib/test/test_cqlsh_completion.py ---------------------------------------------------------------------- diff --git a/pylib/cqlshlib/test/test_cqlsh_completion.py b/pylib/cqlshlib/test/test_cqlsh_completion.py index 90eee4e..fa9490d 100644 --- a/pylib/cqlshlib/test/test_cqlsh_completion.py +++ b/pylib/cqlshlib/test/test_cqlsh_completion.py @@ -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 = " http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/pylib/cqlshlib/test/test_cqlsh_output.py ---------------------------------------------------------------------- diff --git a/pylib/cqlshlib/test/test_cqlsh_output.py b/pylib/cqlshlib/test/test_cqlsh_output.py index b7240f1..2f0d9bf 100644 --- a/pylib/cqlshlib/test/test_cqlsh_output.py +++ b/pylib/cqlshlib/test/test_cqlsh_output.py @@ -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())) http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/auth/AuthKeyspace.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/auth/AuthKeyspace.java b/src/java/org/apache/cassandra/auth/AuthKeyspace.java index a7079dc..9a9dffe 100644 --- a/src/java/org/apache/cassandra/auth/AuthKeyspace.java +++ b/src/java/org/apache/cassandra/auth/AuthKeyspace.java @@ -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(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/cql3/statements/TableAttributes.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/statements/TableAttributes.java b/src/java/org/apache/cassandra/cql3/statements/TableAttributes.java index ba7a001..572362b 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TableAttributes.java +++ b/src/java/org/apache/cassandra/cql3/statements/TableAttributes.java @@ -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))); http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/db/ConsistencyLevel.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java index 19f2e10..840c174 100644 --- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java +++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java @@ -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; @@ -177,18 +175,13 @@ 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 * provide a quorum for each data center. If we are not using a NetworkTopologyStrategy, * 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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/db/SystemKeyspace.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 91a1bff..12ab952 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -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); http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/db/TableCQLHelper.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/TableCQLHelper.java b/src/java/org/apache/cassandra/db/TableCQLHelper.java index 862cf30..0e9977d 100644 --- a/src/java/org/apache/cassandra/db/TableCQLHelper.java +++ b/src/java/org/apache/cassandra/db/TableCQLHelper.java @@ -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())); http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java index 9ee1c60..c79fe89 100644 --- a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java @@ -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")); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java b/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java index b46ae5e..19d83db 100644 --- a/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java +++ b/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java @@ -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(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/schema/SchemaKeyspace.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java index e88f037..ca3c69f 100644 --- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java @@ -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")) http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/schema/TableMetadata.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/schema/TableMetadata.java b/src/java/org/apache/cassandra/schema/TableMetadata.java index 3ccc3c5..4634438 100644 --- a/src/java/org/apache/cassandra/schema/TableMetadata.java +++ b/src/java/org/apache/cassandra/schema/TableMetadata.java @@ -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); http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/schema/TableParams.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/schema/TableParams.java b/src/java/org/apache/cassandra/schema/TableParams.java index b7f6806..78dc894 100644 --- a/src/java/org/apache/cassandra/schema/TableParams.java +++ b/src/java/org/apache/cassandra/schema/TableParams.java @@ -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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java index f45e623..6c8a45a 100644 --- a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java +++ b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java @@ -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) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java b/src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java deleted file mode 100644 index b7e7435..0000000 --- a/src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java +++ /dev/null @@ -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; - } -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/DataResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/DataResolver.java b/src/java/org/apache/cassandra/service/reads/DataResolver.java index 4c7a6c9..ebf6a6f 100644 --- a/src/java/org/apache/cassandra/service/reads/DataResolver.java +++ b/src/java/org/apache/cassandra/service/reads/DataResolver.java @@ -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))); http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/DigestResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/DigestResolver.java b/src/java/org/apache/cassandra/service/reads/DigestResolver.java index 828a65e..b2eb0c6 100644 --- a/src/java/org/apache/cassandra/service/reads/DigestResolver.java +++ b/src/java/org/apache/cassandra/service/reads/DigestResolver.java @@ -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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/ReadCallback.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/ReadCallback.java b/src/java/org/apache/cassandra/service/reads/ReadCallback.java index 62fdfaa..a35fc2e 100644 --- a/src/java/org/apache/cassandra/service/reads/ReadCallback.java +++ b/src/java/org/apache/cassandra/service/reads/ReadCallback.java @@ -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); - } - } } /** http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java b/src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java deleted file mode 100644 index f434c88..0000000 --- a/src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java +++ /dev/null @@ -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; -} http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/ResponseResolver.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/ResponseResolver.java b/src/java/org/apache/cassandra/service/reads/ResponseResolver.java index 69ec063..f4f00a2 100644 --- a/src/java/org/apache/cassandra/service/reads/ResponseResolver.java +++ b/src/java/org/apache/cassandra/service/reads/ResponseResolver.java @@ -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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java index 8689356..f207b7d 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java @@ -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); - } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java index 39f5bff..4436f3a 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java @@ -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) - { - } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java index 21cab20..289875d 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java @@ -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); - - /** - * Wait for any operations started by {@link ReadRepair#startForegroundRepair} to complete - * @throws ReadTimeoutException - */ - 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); + public void startRepair(DigestResolver digestResolver, + List<InetAddressAndPort> allEndpoints, + List<InetAddressAndPort> contactedEndpoints, + Consumer<PartitionIterator> resultConsumer); /** - * If {@link ReadRepair#maybeStartBackgroundRepair} was called with a {@link DigestResolver}, this will - * be called to perform a repair if there was a digest mismatch + * Wait for any operations started by {@link ReadRepair#startRepair} to complete */ - public void backgroundDigestRepair(TraceState traceState); + public void awaitRepair() throws ReadTimeoutException; static ReadRepair create(ReadCommand command, List<InetAddressAndPort> endpoints, long queryStartNanoTime, ConsistencyLevel consistency) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/src/java/org/apache/cassandra/tracing/TraceKeyspace.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/tracing/TraceKeyspace.java b/src/java/org/apache/cassandra/tracing/TraceKeyspace.java index 487ed65..c29760e 100644 --- a/src/java/org/apache/cassandra/tracing/TraceKeyspace.java +++ b/src/java/org/apache/cassandra/tracing/TraceKeyspace.java @@ -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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/OverflowTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/OverflowTest.java b/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/OverflowTest.java index 406f27a..71d632d 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/OverflowTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/miscellaneous/OverflowTest.java @@ -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 } " http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java b/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java index cfc9686..1e465b3 100644 --- a/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java +++ b/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java @@ -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" + http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/test/unit/org/apache/cassandra/schema/MigrationManagerTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/schema/MigrationManagerTest.java b/test/unit/org/apache/cassandra/schema/MigrationManagerTest.java index 6247c02..455fd23 100644 --- a/test/unit/org/apache/cassandra/schema/MigrationManagerTest.java +++ b/test/unit/org/apache/cassandra/schema/MigrationManagerTest.java @@ -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(); } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/2fcd29b8/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java b/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java index 522e524..5664c9b 100644 --- a/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java +++ b/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java @@ -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 { } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
