From 2fcd29b830e7b201e7047d283de385d5f1c427b5 Mon Sep 17 00:00:00 2001 From: Aleksey Yeshchenko Date: Wed, 28 Mar 2018 12:47:09 +0100 Subject: [PATCH] Eliminate background repair and probablistic read_repair_chance table options patch by Aleksey Yeschenko; reviewed by Blake Eggleston for CASSANDRA-13910 --- 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 - .../apache/cassandra/auth/AuthKeyspace.java | 1 - .../cql3/statements/TableAttributes.java | 6 -- .../apache/cassandra/db/ConsistencyLevel.java | 47 ++------------ .../apache/cassandra/db/SystemKeyspace.java | 1 - .../apache/cassandra/db/TableCQLHelper.java | 2 - .../cassandra/metrics/ReadRepairMetrics.java | 3 + .../repair/SystemDistributedKeyspace.java | 1 - .../cassandra/schema/SchemaKeyspace.java | 15 ++--- .../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 ++------- .../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 ++------- .../cassandra/tracing/TraceKeyspace.java | 1 - .../miscellaneous/OverflowTest.java | 4 -- .../cassandra/db/TableCQLHelperTest.java | 4 -- .../schema/MigrationManagerTest.java | 2 - .../reads/repair/TestableReadRepair.java | 18 +----- 36 files changed, 59 insertions(+), 440 deletions(-) delete mode 100644 src/java/org/apache/cassandra/service/reads/AsyncRepairCallback.java delete mode 100644 src/java/org/apache/cassandra/service/reads/ReadRepairDecision.java diff --git a/CHANGES.txt b/CHANGES.txt index f2c9ffca72..76321ddbf9 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) diff --git a/NEWS.txt b/NEWS.txt index 3be52dc464..e8e88a4f42 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 ------------------- diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 1be6feb966..d466072df8 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 diff --git a/doc/cql3/CQL.textile b/doc/cql3/CQL.textile index d87e72a32b..8d55baa831 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 @@: |_. 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@. diff --git a/doc/source/architecture/dynamo.rst b/doc/source/architecture/dynamo.rst index a7dbb87501..365a695cbb 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 ~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/doc/source/cql/ddl.rst b/doc/source/cql/ddl.rst index be5e2c6273..c8cedcf0fc 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 | | | | | `. | +--------------------------------+----------+-------------+-----------------------------------------------------------+ @@ -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: diff --git a/doc/source/operating/compaction.rst b/doc/source/operating/compaction.rst index cb6be458f1..ace9aa9e4a 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 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/doc/source/operating/snitch.rst b/doc/source/operating/snitch.rst index 91bab29394..5f6760a41a 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 diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py index ca14a36f64..314c431d38 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 ["''"] - if this_opt in ('read_repair_chance', 'bloom_filter_fp_chance', - 'dclocal_read_repair_chance'): + if this_opt == 'bloom_filter_fp_chance': return [Hint('')] if this_opt in ('min_compaction_threshold', 'max_compaction_threshold', 'gc_grace_seconds', 'min_index_interval', 'max_index_interval'): diff --git a/pylib/cqlshlib/test/test_cqlsh_completion.py b/pylib/cqlshlib/test/test_cqlsh_completion.py index 90eee4efde..fa9490dbe8 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 = " diff --git a/pylib/cqlshlib/test/test_cqlsh_output.py b/pylib/cqlshlib/test/test_cqlsh_output.py index b7240f10d0..2f0d9bfb8a 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())) diff --git a/src/java/org/apache/cassandra/auth/AuthKeyspace.java b/src/java/org/apache/cassandra/auth/AuthKeyspace.java index a7079dc6ae..9a9dffea64 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(); } diff --git a/src/java/org/apache/cassandra/cql3/statements/TableAttributes.java b/src/java/org/apache/cassandra/cql3/statements/TableAttributes.java index ba7a001a9d..572362b810 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))); diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java index 19f2e100b1..840c174b6b 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; @@ -176,11 +174,6 @@ public enum ConsistencyLevel } public List filterForQuery(Keyspace keyspace, List liveEndpoints) - { - return filterForQuery(keyspace, liveEndpoints, ReadRepairDecision.NONE); - } - - public List filterForQuery(Keyspace keyspace, List 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 local = new ArrayList<>(); - List 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 filterForEachQuorum(Keyspace keyspace, List liveEndpoints, ReadRepairDecision readRepair) + private List filterForEachQuorum(Keyspace keyspace, List 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> dcsEndpoints = new HashMap<>(); for (String dc: strategy.getDatacenters()) dcsEndpoints.put(dc, new ArrayList<>()); @@ -247,10 +213,7 @@ public enum ConsistencyLevel for (Map.Entry> dcEndpoints : dcsEndpoints.entrySet()) { List 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; diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index 91a1bffc97..12ab9528aa 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); diff --git a/src/java/org/apache/cassandra/db/TableCQLHelper.java b/src/java/org/apache/cassandra/db/TableCQLHelper.java index 862cf30331..0e9977d404 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())); diff --git a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java index 9ee1c6073c..c79fe895d7 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")); } diff --git a/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java b/src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java index b46ae5ea3f..19d83dbfe6 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(); } diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java index e88f03717a..ca3c69f066 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>," + "compression frozen>," + "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>," + "flags frozen>," // 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>," + "compression frozen>," + "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>," + "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")) diff --git a/src/java/org/apache/cassandra/schema/TableMetadata.java b/src/java/org/apache/cassandra/schema/TableMetadata.java index 3ccc3c5b62..4634438e46 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); diff --git a/src/java/org/apache/cassandra/schema/TableParams.java b/src/java/org/apache/cassandra/schema/TableParams.java index b7f6806b54..78dc8943ee 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; diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java index f45e6235fe..6c8a45a3dd 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 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 targetReplicas = consistencyLevel.filterForQuery(keyspace, allReplicas, repairDecision); + List 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) { 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 b7e7435f4b..0000000000 --- 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 -{ - 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 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; - } -} diff --git a/src/java/org/apache/cassandra/service/reads/DataResolver.java b/src/java/org/apache/cassandra/service/reads/DataResolver.java index 4c7a6c9f22..ebf6a6f456 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))); diff --git a/src/java/org/apache/cassandra/service/reads/DigestResolver.java b/src/java/org/apache/cassandra/service/reads/DigestResolver.java index 828a65e4f8..b2eb0c6ba9 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; diff --git a/src/java/org/apache/cassandra/service/reads/ReadCallback.java b/src/java/org/apache/cassandra/service/reads/ReadCallback.java index 62fdfaa984..a35fc2e897 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 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); - } - } } /** 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 f434c88e02..0000000000 --- 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; -} diff --git a/src/java/org/apache/cassandra/service/reads/ResponseResolver.java b/src/java/org/apache/cassandra/service/reads/ResponseResolver.java index 69ec063e07..f4f00a2043 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 message) 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 868935665d..f207b7d01b 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 allEndpoints, List contactedEndpoints, Consumer resultConsumer) + public void startRepair(DigestResolver digestResolver, List allEndpoints, List contactedEndpoints, Consumer 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); - } } 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 39f5bff52b..4436f3a038 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 allEndpoints, List contactedEndpoints, Consumer resultConsumer) + public void startRepair(DigestResolver digestResolver, List allEndpoints, List contactedEndpoints, Consumer 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) - { - } } 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 21cab20df6..289875db20 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 allEndpoints, - List contactedEndpoints, - Consumer resultConsumer); + public void startRepair(DigestResolver digestResolver, + List allEndpoints, + List contactedEndpoints, + Consumer 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 endpoints, long queryStartNanoTime, ConsistencyLevel consistency) { diff --git a/src/java/org/apache/cassandra/tracing/TraceKeyspace.java b/src/java/org/apache/cassandra/tracing/TraceKeyspace.java index 487ed6578f..c29760e35a 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) 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 406f27a8dc..71d632dee0 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 } " diff --git a/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java b/test/unit/org/apache/cassandra/db/TableCQLHelperTest.java index cfc96869bd..1e465b3e1e 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" + diff --git a/test/unit/org/apache/cassandra/schema/MigrationManagerTest.java b/test/unit/org/apache/cassandra/schema/MigrationManagerTest.java index 6247c02be1..455fd23a28 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(); } } 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 522e524bdc..5664c9b497 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 allEndpoints, List contactedEndpoints, Consumer resultConsumer) + public void startRepair(DigestResolver digestResolver, List allEndpoints, List contactedEndpoints, Consumer resultConsumer) { } @Override - public void awaitForegroundRepairFinish() throws ReadTimeoutException - { - - } - - @Override - public void maybeStartBackgroundRepair(ResponseResolver resolver) - { - - } - - @Override - public void backgroundDigestRepair(TraceState traceState) + public void awaitRepair() throws ReadTimeoutException { }