| /* |
| * 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.distributed.test; |
| |
| import java.io.IOException; |
| import java.util.*; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CyclicBarrier; |
| import java.util.concurrent.TimeUnit; |
| import java.util.stream.IntStream; |
| import java.util.stream.Stream; |
| |
| import com.google.common.util.concurrent.Uninterruptibles; |
| import org.apache.cassandra.concurrent.SEPExecutor; |
| import org.apache.cassandra.dht.Token; |
| import org.apache.cassandra.locator.AbstractReplicationStrategy; |
| import org.apache.cassandra.locator.EndpointsForToken; |
| import org.apache.cassandra.locator.InetAddressAndPort; |
| import org.apache.cassandra.locator.ReplicaLayout; |
| import org.apache.cassandra.locator.ReplicaUtils; |
| import org.apache.cassandra.utils.Throwables; |
| import org.junit.Assert; |
| import org.junit.Test; |
| |
| import net.bytebuddy.ByteBuddy; |
| import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; |
| import net.bytebuddy.implementation.MethodDelegation; |
| import net.bytebuddy.implementation.bind.annotation.SuperCall; |
| import org.apache.cassandra.db.*; |
| import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; |
| import org.apache.cassandra.distributed.Cluster; |
| import org.apache.cassandra.distributed.api.ConsistencyLevel; |
| import org.apache.cassandra.distributed.api.IInvokableInstance; |
| import org.apache.cassandra.distributed.api.IIsolatedExecutor; |
| import org.apache.cassandra.io.sstable.Descriptor; |
| import org.apache.cassandra.io.sstable.format.SSTableReader; |
| import org.apache.cassandra.io.sstable.metadata.MetadataComponent; |
| import org.apache.cassandra.io.sstable.metadata.MetadataType; |
| import org.apache.cassandra.io.sstable.metadata.StatsMetadata; |
| import org.apache.cassandra.service.ActiveRepairService; |
| import org.apache.cassandra.service.StorageProxy; |
| import org.apache.cassandra.service.StorageProxy.LocalReadRunnable; |
| import org.apache.cassandra.utils.DiagnosticSnapshotService; |
| |
| import static net.bytebuddy.matcher.ElementMatchers.named; |
| import static net.bytebuddy.matcher.ElementMatchers.takesArguments; |
| import static org.apache.cassandra.distributed.api.Feature.GOSSIP; |
| import static org.apache.cassandra.distributed.api.Feature.NETWORK; |
| import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| public class RepairDigestTrackingTest extends TestBaseImpl |
| { |
| private static final String TABLE = "tbl"; |
| private static final String KS_TABLE = KEYSPACE + '.' + TABLE; |
| |
| @SuppressWarnings("Convert2MethodRef") |
| @Test |
| public void testInconsistenciesFound() throws Throwable |
| { |
| try (Cluster cluster = init(builder().withNodes(2).start())) |
| { |
| |
| cluster.get(1).runOnInstance(() -> StorageProxy.instance.enableRepairedDataTrackingForRangeReads()); |
| |
| cluster.schemaChange("CREATE TABLE " + KS_TABLE+ " (k INT, c INT, v INT, PRIMARY KEY (k,c)) with read_repair='NONE'"); |
| for (int i = 0; i < 10; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v) VALUES (?, ?, ?)", |
| ConsistencyLevel.ALL, |
| i, i, i); |
| } |
| cluster.forEach(i -> i.flush(KEYSPACE)); |
| |
| for (int i = 10; i < 20; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v) VALUES (?, ?, ?)", |
| ConsistencyLevel.ALL, |
| i, i, i); |
| } |
| cluster.forEach(i -> i.flush(KEYSPACE)); |
| cluster.forEach(i -> i.runOnInstance(assertNotRepaired())); |
| |
| // mark everything on node 2 repaired |
| cluster.get(2).runOnInstance(markAllRepaired()); |
| cluster.get(2).runOnInstance(assertRepaired()); |
| |
| // insert more data on node1 to generate an initial mismatch |
| cluster.get(1).executeInternal("INSERT INTO " + KS_TABLE + " (k, c, v) VALUES (?, ?, ?)", 5, 5, 55); |
| cluster.get(1).runOnInstance(assertNotRepaired()); |
| |
| long ccBefore = getConfirmedInconsistencies(cluster.get(1)); |
| cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE, ConsistencyLevel.ALL); |
| long ccAfter = getConfirmedInconsistencies(cluster.get(1)); |
| Assert.assertEquals("confirmed count should differ by 1 after range read", ccBefore + 1, ccAfter); |
| } |
| } |
| |
| @SuppressWarnings("Convert2MethodRef") |
| @Test |
| public void testPurgeableTombstonesAreIgnored() throws Throwable |
| { |
| try (Cluster cluster = init(builder().withNodes(2).start())) |
| { |
| cluster.get(1).runOnInstance(() -> StorageProxy.instance.enableRepairedDataTrackingForRangeReads()); |
| |
| cluster.schemaChange("CREATE TABLE " + KS_TABLE + " (k INT, c INT, v1 INT, v2 INT, PRIMARY KEY (k,c)) WITH gc_grace_seconds=0"); |
| // on node1 only insert some tombstones, then flush |
| for (int i = 0; i < 10; i++) |
| { |
| cluster.get(1).executeInternal("DELETE v1 FROM " + KS_TABLE + " USING TIMESTAMP 0 WHERE k=? and c=? ", i, i); |
| } |
| cluster.get(1).flush(KEYSPACE); |
| |
| // insert data on both nodes and flush |
| for (int i = 0; i < 10; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v2) VALUES (?, ?, ?) USING TIMESTAMP 1", |
| ConsistencyLevel.ALL, |
| i, i, i); |
| } |
| cluster.forEach(i -> i.flush(KEYSPACE)); |
| |
| // nothing is repaired yet |
| cluster.forEach(i -> i.runOnInstance(assertNotRepaired())); |
| // mark everything repaired |
| cluster.forEach(i -> i.runOnInstance(markAllRepaired())); |
| cluster.forEach(i -> i.runOnInstance(assertRepaired())); |
| |
| // now overwrite on node2 only to generate digest mismatches, but don't flush so the repaired dataset is not affected |
| for (int i = 0; i < 10; i++) |
| { |
| cluster.get(2).executeInternal("INSERT INTO " + KS_TABLE + " (k, c, v2) VALUES (?, ?, ?) USING TIMESTAMP 2", i, i, i * 2); |
| } |
| |
| long ccBefore = getConfirmedInconsistencies(cluster.get(1)); |
| // Unfortunately we need to sleep here to ensure that nowInSec > the local deletion time of the tombstones |
| TimeUnit.SECONDS.sleep(2); |
| cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE, ConsistencyLevel.ALL); |
| long ccAfter = getConfirmedInconsistencies(cluster.get(1)); |
| |
| Assert.assertEquals("No repaired data inconsistencies should be detected", ccBefore, ccAfter); |
| } |
| } |
| |
| @SuppressWarnings("Convert2MethodRef") |
| @Test |
| public void testSnapshottingOnInconsistency() throws Throwable |
| { |
| try (Cluster cluster = init(Cluster.create(2))) |
| { |
| cluster.get(1).runOnInstance(() -> StorageProxy.instance.enableRepairedDataTrackingForPartitionReads()); |
| |
| cluster.schemaChange("CREATE TABLE " + KS_TABLE + " (k INT, c INT, v INT, PRIMARY KEY (k,c))"); |
| for (int i = 0; i < 10; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v) VALUES (0, ?, ?)", |
| ConsistencyLevel.ALL, i, i); |
| } |
| cluster.forEach(c -> c.flush(KEYSPACE)); |
| |
| for (int i = 10; i < 20; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v) VALUES (0, ?, ?)", |
| ConsistencyLevel.ALL, i, i); |
| } |
| cluster.forEach(c -> c.flush(KEYSPACE)); |
| cluster.forEach(i -> i.runOnInstance(assertNotRepaired())); |
| // Mark everything repaired on node2 |
| cluster.get(2).runOnInstance(markAllRepaired()); |
| cluster.get(2).runOnInstance(assertRepaired()); |
| |
| // now overwrite on node1 only to generate digest mismatches |
| cluster.get(1).executeInternal("INSERT INTO " + KS_TABLE + " (k, c, v) VALUES (0, ?, ?)", 5, 55); |
| cluster.get(1).runOnInstance(assertNotRepaired()); |
| |
| // Execute a partition read and assert inconsistency is detected (as nothing is repaired on node1) |
| long ccBefore = getConfirmedInconsistencies(cluster.get(1)); |
| cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE + " WHERE k=0", ConsistencyLevel.ALL); |
| long ccAfter = getConfirmedInconsistencies(cluster.get(1)); |
| Assert.assertEquals("confirmed count should increment by 1 after each partition read", ccBefore + 1, ccAfter); |
| |
| String snapshotName = DiagnosticSnapshotService.getSnapshotName(DiagnosticSnapshotService.REPAIRED_DATA_MISMATCH_SNAPSHOT_PREFIX); |
| |
| cluster.forEach(i -> i.runOnInstance(assertSnapshotNotPresent(snapshotName))); |
| |
| // re-introduce a mismatch, enable snapshotting and try again |
| cluster.get(1).executeInternal("INSERT INTO " + KS_TABLE + " (k, c, v) VALUES (0, ?, ?)", 5, 555); |
| cluster.get(1).runOnInstance(() -> StorageProxy.instance.enableSnapshotOnRepairedDataMismatch()); |
| |
| cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE + " WHERE k=0", ConsistencyLevel.ALL); |
| ccAfter = getConfirmedInconsistencies(cluster.get(1)); |
| Assert.assertEquals("confirmed count should increment by 1 after each partition read", ccBefore + 2, ccAfter); |
| |
| cluster.forEach(i -> i.runOnInstance(assertSnapshotPresent(snapshotName))); |
| } |
| } |
| |
| @Test |
| public void testRepairedReadCountNormalizationWithInitialUnderread() throws Throwable |
| { |
| // Asserts that the amount of repaired data read for digest generation is consistent |
| // across replicas where one has to read less repaired data to satisfy the original |
| // limits of the read request. |
| try (Cluster cluster = init(Cluster.create(2))) |
| { |
| |
| cluster.get(1).runOnInstance(() -> { |
| StorageProxy.instance.enableRepairedDataTrackingForRangeReads(); |
| StorageProxy.instance.enableRepairedDataTrackingForPartitionReads(); |
| }); |
| |
| cluster.schemaChange("CREATE TABLE " + KS_TABLE + " (k INT, c INT, v1 INT, PRIMARY KEY (k,c)) " + |
| "WITH CLUSTERING ORDER BY (c DESC)"); |
| |
| // insert data on both nodes and flush |
| for (int i=0; i<20; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (0, ?, ?) USING TIMESTAMP 0", |
| ConsistencyLevel.ALL, i, i); |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (1, ?, ?) USING TIMESTAMP 1", |
| ConsistencyLevel.ALL, i, i); |
| } |
| cluster.forEach(c -> c.flush(KEYSPACE)); |
| // nothing is repaired yet |
| cluster.forEach(i -> i.runOnInstance(assertNotRepaired())); |
| // mark everything repaired |
| cluster.forEach(i -> i.runOnInstance(markAllRepaired())); |
| cluster.forEach(i -> i.runOnInstance(assertRepaired())); |
| |
| // Add some unrepaired data to both nodes |
| for (int i=20; i<30; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (1, ?, ?) USING TIMESTAMP 1", |
| ConsistencyLevel.ALL, i, i); |
| } |
| // And some more unrepaired data to node2 only. This causes node2 to read less repaired data than node1 |
| // when satisfying the limits of the read. So node2 needs to overread more repaired data than node1 when |
| // calculating the repaired data digest. |
| cluster.get(2).executeInternal("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (1, ?, ?) USING TIMESTAMP 1", 30, 30); |
| |
| // Verify single partition read |
| long ccBefore = getConfirmedInconsistencies(cluster.get(1)); |
| assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE + " WHERE k=1 LIMIT 20", ConsistencyLevel.ALL), |
| rows(1, 30, 11)); |
| long ccAfterPartitionRead = getConfirmedInconsistencies(cluster.get(1)); |
| |
| // Recreate a mismatch in unrepaired data and verify partition range read |
| cluster.get(2).executeInternal("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (1, ?, ?)", 31, 31); |
| assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE + " LIMIT 30", ConsistencyLevel.ALL), |
| rows(1, 31, 2)); |
| long ccAfterRangeRead = getConfirmedInconsistencies(cluster.get(1)); |
| |
| if (ccAfterPartitionRead != ccAfterRangeRead) |
| if (ccAfterPartitionRead != ccBefore) |
| fail("Both range and partition reads reported data inconsistencies but none were expected"); |
| else |
| fail("Reported inconsistency during range read but none were expected"); |
| else if (ccAfterPartitionRead != ccBefore) |
| fail("Reported inconsistency during partition read but none were expected"); |
| } |
| } |
| |
| @Test |
| public void testRepairedReadCountNormalizationWithInitialOverread() throws Throwable |
| { |
| // Asserts that the amount of repaired data read for digest generation is consistent |
| // across replicas where one has to read more repaired data to satisfy the original |
| // limits of the read request. |
| try (Cluster cluster = init(Cluster.create(2))) |
| { |
| |
| cluster.get(1).runOnInstance(() -> { |
| StorageProxy.instance.enableRepairedDataTrackingForRangeReads(); |
| StorageProxy.instance.enableRepairedDataTrackingForPartitionReads(); |
| }); |
| |
| cluster.schemaChange("CREATE TABLE " + KS_TABLE + " (k INT, c INT, v1 INT, PRIMARY KEY (k,c)) " + |
| "WITH CLUSTERING ORDER BY (c DESC)"); |
| |
| // insert data on both nodes and flush |
| for (int i=0; i<10; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (0, ?, ?) USING TIMESTAMP 0", |
| ConsistencyLevel.ALL, i, i); |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (1, ?, ?) USING TIMESTAMP 1", |
| ConsistencyLevel.ALL, i, i); |
| } |
| cluster.forEach(c -> c.flush(KEYSPACE)); |
| // nothing is repaired yet |
| cluster.forEach(i -> i.runOnInstance(assertNotRepaired())); |
| // mark everything repaired |
| cluster.forEach(i -> i.runOnInstance(markAllRepaired())); |
| cluster.forEach(i -> i.runOnInstance(assertRepaired())); |
| |
| // Add some unrepaired data to both nodes |
| for (int i=10; i<13; i++) |
| { |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (0, ?, ?) USING TIMESTAMP 1", |
| ConsistencyLevel.ALL, i, i); |
| cluster.coordinator(1).execute("INSERT INTO " + KS_TABLE + " (k, c, v1) VALUES (1, ?, ?) USING TIMESTAMP 1", |
| ConsistencyLevel.ALL, i, i); |
| } |
| cluster.forEach(c -> c.flush(KEYSPACE)); |
| // And some row deletions on node2 only which cover data in the repaired set |
| // This will cause node2 to read more repaired data in satisfying the limit of the read request |
| // so it should overread less than node1 (in fact, it should not overread at all) in order to |
| // calculate the repaired data digest. |
| for (int i=7; i<10; i++) |
| { |
| cluster.get(2).executeInternal("DELETE FROM " + KS_TABLE + " USING TIMESTAMP 2 WHERE k = 0 AND c = ?", i); |
| cluster.get(2).executeInternal("DELETE FROM " + KS_TABLE + " USING TIMESTAMP 2 WHERE k = 1 AND c = ?", i); |
| } |
| |
| // Verify single partition read |
| long ccBefore = getConfirmedInconsistencies(cluster.get(1)); |
| assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE + " WHERE k=0 LIMIT 5", ConsistencyLevel.ALL), |
| rows(rows(0, 12, 10), rows(0, 6, 5))); |
| long ccAfterPartitionRead = getConfirmedInconsistencies(cluster.get(1)); |
| |
| // Verify partition range read |
| assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE + " LIMIT 11", ConsistencyLevel.ALL), |
| rows(rows(1, 12, 10), rows(1, 6, 0), rows(0, 12, 12))); |
| long ccAfterRangeRead = getConfirmedInconsistencies(cluster.get(1)); |
| |
| if (ccAfterPartitionRead != ccAfterRangeRead) |
| if (ccAfterPartitionRead != ccBefore) |
| fail("Both range and partition reads reported data inconsistencies but none were expected"); |
| else |
| fail("Reported inconsistency during range read but none were expected"); |
| else if (ccAfterPartitionRead != ccBefore) |
| fail("Reported inconsistency during partition read but none were expected"); |
| } |
| } |
| |
| /** |
| * In CASSANDRA-16721, we discovered that if responses from remote replicas came back while the local runnable was |
| * still executing, the fact that {@link ReadCommand} was mutable meant that the trackRepairedStatus flag on the |
| * command instance could move from false to true in executeLocally(), between setting the |
| * RepairedDataInfo/gathering the sstables and calling extend(). When this happened, the RDI was still the |
| * stand-in object NO_OP_REPAIRED_DATA_INFO, which has a null repairedDataCounter, and we hit the NPE. |
| * |
| * Similarly, the trackRepairedStatus flag could be set after the point at which the RDI is set on the local |
| * read, assigned to the repairedDataInfo in {@link ReadCommand}, and improperly shared between initial local read |
| * and the local read triggered by read repair. |
| * |
| * These problems are sidestepped completely by CASSANDRA-16721, as an RDI instance is now created and destroyed |
| * entirely within the scope of single {@link LocalReadRunnable}, but this test still attempts to validate some |
| * assumptions about the cleanliness of the logs and the correctness of queries made when initial local reads and |
| * local reads triggered by read repair (after speculative reads) execute at roughly the same time. |
| * |
| * This test depends on whether node1 gets a data or a digest request first, we force it to be a digest request |
| * in the forTokenReadLiveSorted ByteBuddy rule below. |
| */ |
| @Test |
| public void testLocalDataAndRemoteRequestConcurrency() throws Exception |
| { |
| try (Cluster cluster = init(Cluster.build(3) |
| .withInstanceInitializer(BBHelper::install) |
| .withConfig(config -> config.set("repaired_data_tracking_for_partition_reads_enabled", true) |
| .with(GOSSIP) |
| .with(NETWORK)) |
| .start())) |
| { |
| // A speculative read is the reason we have two remote replicas in play that can return results before |
| // the local replica does. |
| setupSchema(cluster, "create table " + KS_TABLE + " (id int primary key, t int) WITH speculative_retry = 'ALWAYS'"); |
| |
| cluster.get(1).executeInternal("INSERT INTO " + KS_TABLE + " (id, t) values (0, 0)"); |
| cluster.get(2).executeInternal("INSERT INTO " + KS_TABLE + " (id, t) values (0, 0)"); |
| cluster.get(3).executeInternal("INSERT INTO " + KS_TABLE + " (id, t) values (0, 1)"); |
| cluster.forEach(c -> c.flush(KEYSPACE)); |
| cluster.forEach(i -> i.runOnInstance(markAllRepaired())); |
| cluster.forEach(i -> i.runOnInstance(assertRepaired())); |
| |
| long logPositionBeforeQuery = cluster.get(1).logs().mark(); |
| Object[][] rows = cluster.coordinator(1).execute("SELECT * FROM " + KS_TABLE + " WHERE id=0", ConsistencyLevel.QUORUM); |
| assertEquals(1, rows.length); |
| |
| // Given we didn't write at QUORUM, both 0 and 1 are acceptable values. |
| assertTrue((int) rows[0][1] == 0 || (int) rows[0][1] == 1); |
| |
| List<String> result = cluster.get(1).logs().grepForErrors(logPositionBeforeQuery).getResult(); |
| assertEquals(Collections.emptyList(), result); |
| Assert.assertTrue("Encountered an error", result.isEmpty()); |
| } |
| } |
| |
| public static class BBHelper |
| { |
| private static final CyclicBarrier barrier = new CyclicBarrier(2); |
| |
| public static void install(ClassLoader classLoader, Integer num) |
| { |
| // Only install on the coordinating node, which is also a replica... |
| if (num == 1) |
| { |
| new ByteBuddy().rebase(SEPExecutor.class) |
| .method(named("maybeExecuteImmediately")) |
| .intercept(MethodDelegation.to(BBHelper.class)) |
| .make() |
| .load(classLoader, ClassLoadingStrategy.Default.INJECTION); |
| |
| new ByteBuddy().rebase(SinglePartitionReadCommand.class) |
| .method(named("executeLocally")) |
| .intercept(MethodDelegation.to(BBHelper.class)) |
| .make() |
| .load(classLoader, ClassLoadingStrategy.Default.INJECTION); |
| |
| new ByteBuddy().rebase(ReplicaLayout.class) |
| .method(named("forTokenReadLiveSorted").and(takesArguments(AbstractReplicationStrategy.class, Token.class))) |
| .intercept(MethodDelegation.to(BBHelper.class)) |
| .make() |
| .load(classLoader, ClassLoadingStrategy.Default.INJECTION); |
| } |
| } |
| |
| @SuppressWarnings("unused") |
| public static void maybeExecuteImmediately(Runnable command) |
| { |
| // Force local read runnables (from initial read and read-repair) to execute in separate threads. |
| new Thread(command).start(); |
| } |
| |
| @SuppressWarnings({ "unused" }) |
| public static UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController, |
| @SuperCall Callable<UnfilteredPartitionIterator> zuperCall) |
| { |
| try |
| { |
| if (executionController.metadata().name.equals(TABLE)) |
| { |
| // Force both the initial local read and the local read triggered by read-repair to proceed at |
| // roughly the same time. |
| barrier.await(); |
| } |
| return zuperCall.call(); |
| } |
| catch (Exception e) |
| { |
| throw Throwables.unchecked(e); |
| } |
| } |
| |
| @SuppressWarnings({ "unused" }) |
| public static ReplicaLayout.ForTokenRead forTokenReadLiveSorted(AbstractReplicationStrategy replicationStrategy, Token token) |
| { |
| try |
| { |
| EndpointsForToken.Builder builder = EndpointsForToken.builder(token, 3); |
| builder.add(ReplicaUtils.full(InetAddressAndPort.getByName("127.0.0.3"))); |
| builder.add(ReplicaUtils.full(InetAddressAndPort.getByName("127.0.0.2"))); |
| builder.add(ReplicaUtils.full(InetAddressAndPort.getByName("127.0.0.1"))); |
| return new ReplicaLayout.ForTokenRead(replicationStrategy, builder.build()); |
| } |
| catch (Exception e) |
| { |
| throw Throwables.unchecked(e); |
| } |
| } |
| } |
| |
| private Object[][] rows(Object[][] head, Object[][]...tail) |
| { |
| return Stream.concat(Stream.of(head), |
| Stream.of(tail).flatMap(Stream::of)) |
| .toArray(Object[][]::new); |
| } |
| |
| private Object[][] rows(int partitionKey, int start, int end) |
| { |
| if (start == end) |
| return new Object[][] { new Object[] { partitionKey, start, end } }; |
| |
| IntStream clusterings = start > end |
| ? IntStream.range(end -1, start).map(i -> start - i + end - 1) |
| : IntStream.range(start, end); |
| |
| return clusterings.mapToObj(i -> new Object[] {partitionKey, i, i}).toArray(Object[][]::new); |
| } |
| |
| private IIsolatedExecutor.SerializableRunnable assertNotRepaired() |
| { |
| return () -> |
| { |
| try |
| { |
| Iterator<SSTableReader> sstables = Keyspace.open(KEYSPACE) |
| .getColumnFamilyStore(TABLE) |
| .getLiveSSTables() |
| .iterator(); |
| while (sstables.hasNext()) |
| { |
| SSTableReader sstable = sstables.next(); |
| Descriptor descriptor = sstable.descriptor; |
| Map<MetadataType, MetadataComponent> metadata = descriptor.getMetadataSerializer() |
| .deserialize(descriptor, EnumSet.of(MetadataType.STATS)); |
| |
| StatsMetadata stats = (StatsMetadata) metadata.get(MetadataType.STATS); |
| Assert.assertEquals("repaired at is set for sstable: " + descriptor, |
| stats.repairedAt, |
| ActiveRepairService.UNREPAIRED_SSTABLE); |
| } |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| }; |
| } |
| |
| private IIsolatedExecutor.SerializableRunnable markAllRepaired() |
| { |
| return () -> |
| { |
| try |
| { |
| Iterator<SSTableReader> sstables = Keyspace.open(KEYSPACE) |
| .getColumnFamilyStore(TABLE) |
| .getLiveSSTables() |
| .iterator(); |
| while (sstables.hasNext()) |
| { |
| SSTableReader sstable = sstables.next(); |
| Descriptor descriptor = sstable.descriptor; |
| descriptor.getMetadataSerializer() |
| .mutateRepairMetadata(descriptor, System.currentTimeMillis(), null, false); |
| sstable.reloadSSTableMetadata(); |
| } |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| }; |
| } |
| |
| private IIsolatedExecutor.SerializableRunnable assertRepaired() |
| { |
| return () -> |
| { |
| try |
| { |
| Iterator<SSTableReader> sstables = Keyspace.open(KEYSPACE) |
| .getColumnFamilyStore(TABLE) |
| .getLiveSSTables() |
| .iterator(); |
| while (sstables.hasNext()) |
| { |
| SSTableReader sstable = sstables.next(); |
| Descriptor descriptor = sstable.descriptor; |
| Map<MetadataType, MetadataComponent> metadata = descriptor.getMetadataSerializer() |
| .deserialize(descriptor, EnumSet.of(MetadataType.STATS)); |
| |
| StatsMetadata stats = (StatsMetadata) metadata.get(MetadataType.STATS); |
| Assert.assertTrue("repaired at is not set for sstable: " + descriptor, stats.repairedAt > 0); |
| } |
| } |
| catch (IOException e) |
| { |
| throw new RuntimeException(e); |
| } |
| }; |
| } |
| |
| @SuppressWarnings("UnstableApiUsage") |
| private IInvokableInstance.SerializableRunnable assertSnapshotPresent(String snapshotName) |
| { |
| return () -> |
| { |
| // snapshots are taken asynchronously, this is crude but it gives it a chance to happen |
| int attempts = 100; |
| ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE); |
| |
| while (cfs.getSnapshotDetails().isEmpty()) |
| { |
| Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); |
| if (attempts-- < 0) |
| throw new AssertionError(String.format("Snapshot %s not found for for %s", snapshotName, KS_TABLE)); |
| } |
| }; |
| } |
| |
| private IInvokableInstance.SerializableRunnable assertSnapshotNotPresent(String snapshotName) |
| { |
| return () -> |
| { |
| ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE); |
| Assert.assertFalse(cfs.snapshotExists(snapshotName)); |
| }; |
| } |
| |
| private long getConfirmedInconsistencies(IInvokableInstance instance) |
| { |
| return instance.callOnInstance(() -> Keyspace.open(KEYSPACE) |
| .getColumnFamilyStore(TABLE) |
| .metric |
| .confirmedRepairedInconsistencies |
| .table |
| .getCount()); |
| } |
| |
| private void setupSchema(Cluster cluster, String cql) |
| { |
| cluster.schemaChange(cql); |
| // disable auto compaction to prevent nodes from trying to compact |
| // new sstables with ones we've modified to mark repaired |
| cluster.forEach(i -> i.runOnInstance(() -> Keyspace.open(KEYSPACE) |
| .getColumnFamilyStore(TABLE) |
| .disableAutoCompaction())); |
| } |
| } |
