diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedPreviewRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedPreviewRepairTest.java new file mode 100644 index 000000000000..8ed6d5b18c3b --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedPreviewRepairTest.java @@ -0,0 +1,656 @@ +/* + * 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.tracking; + +import java.io.IOException; +import java.util.List; +import java.util.StringJoiner; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; + +import javax.management.Notification; +import javax.management.NotificationListener; + +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInstanceConfig; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.NodeToolResult; +import org.apache.cassandra.distributed.shared.Uninterruptibles; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.metrics.RepairMetrics; +import org.apache.cassandra.replication.MutationTrackingService; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.service.snapshot.SnapshotManager; +import org.apache.cassandra.service.snapshot.TableSnapshot; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.utils.DiagnosticSnapshotService; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class TrackedPreviewRepairTest extends TestBaseImpl +{ + private static final int NODES = 3; + + private static final Consumer CONFIG = cfg -> cfg.with(Feature.NETWORK) + .with(Feature.GOSSIP) + .set("snapshot_on_repaired_data_mismatch", true); + + private static Cluster CLUSTER; + private static final AtomicInteger UNIQUE_SUFFIX = new AtomicInteger(); + + private String keyspace; + private String table; + + @BeforeClass + public static void setupCluster() throws IOException + { + CLUSTER = Cluster.build(NODES).withConfig(CONFIG).start(); + } + + @AfterClass + public static void teardownCluster() + { + if (CLUSTER != null) + { + CLUSTER.close(); + CLUSTER = null; + } + } + + @Before + public void assignUniqueNames() + { + int suffix = UNIQUE_SUFFIX.incrementAndGet(); + keyspace = "preview_ks_" + suffix; + table = "tbl_" + suffix; + } + + /** + * On a fully-migrated tracked keyspace with all data reconciled, + * `nodetool repair --validate` succeeds via the MT preview path (not classic) + * and reports "in sync" with no metric increment or diagnostic snapshot. + *

+ * The routing proof is the presence of "establishing cut" and "audit complete" + * progress notifications, which are unique to MutationTrackingRepairedPreviewTask. + */ + @Test + public void previewOnFullyReconciledTrackedKeyspaceReportsInSyncAndRoutesToTrackedPath() + { + createTrackedSchema(CLUSTER, keyspace, table); + + for (int i = 0; i < 10; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + settleReconciliation(CLUSTER); + + long[] previewFailuresBefore = previewFailuresPerNode(CLUSTER); + + NodeToolResult result = CLUSTER.get(1).nodetoolResult(true, "repair", "--validate", keyspace); + result.asserts().success(); + + // Routing proof: distinctive progress notifications only the MT preview task emits. + assertNotificationContains(result, "establishing cut"); + assertNotificationContains(result, "audit complete"); + + // Reused wording from classic PreviewRepairTask so operators see one output shape. + assertNotificationContains(result, "Repaired data is in sync"); + + // In-sync run must not bump the previewFailures counter on any node. + assertPreviewFailuresUnchanged(CLUSTER, previewFailuresBefore); + + // In-sync run must not take a diagnostic snapshot on any replica. + assertNoDiagnosticSnapshot(CLUSTER, keyspace, table); + } + + /** + * Writes issued after the audit's cut is established must not be included in + * the audit tree. The audit is started asynchronously, then paused (from the + * driver's perspective) until the coordinator emits the "cut established" + * progress notification; additional writes are then issued at CL.ALL before + * the audit is allowed to run to completion. The audit should still report + * "in sync" because those post-cut writes are excluded from validation. + */ + @Test + public void previewIgnoresWritesIssuedAfterCutIsEstablished() throws Exception + { + createTrackedSchema(CLUSTER, keyspace, table); + + for (int i = 0; i < 10; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + settleReconciliation(CLUSTER); + + long[] previewFailuresBefore = previewFailuresPerNode(CLUSTER); + + IInvokableInstance coordinator = CLUSTER.get(1); + coordinator.runOnInstance(CutEstablishedLatch::install); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + + try + { + Future resultFuture = executor.submit(() -> coordinator.nodetoolResult(true, "repair", "--validate", keyspace)); + + // Block until the "cut established" progress notification fires on the coordinator. + coordinator.runOnInstance(CutEstablishedLatch::await); + + /* + * TODO: this only synchronizes when the driver starts writing — the audit + * itself is not blocked, so writes race with validation. Once the production + * MT_AUDIT_VALIDATE_REQ verb exists, replace CutEstablishedLatch with an + * outbound message filter on that verb whose matcher blocks the sender until + * released, so the audit is truly held between cut establishment and validation. + */ + + // These writes must land at offsets > cut and be excluded from the audit. + for (int i = 10; i < 20; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + NodeToolResult result = resultFuture.get(120, TimeUnit.SECONDS); + result.asserts().success(); + + assertNotificationContains(result, "cut established"); + assertNotificationContains(result, "audit complete"); + assertNotificationContains(result, "Repaired data is in sync"); + + assertPreviewFailuresUnchanged(CLUSTER, previewFailuresBefore); + assertNoDiagnosticSnapshot(CLUSTER, keyspace, table); + } + finally + { + coordinator.runOnInstance(CutEstablishedLatch::uninstall); + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + /** + * Divergence in an already-reconciled SSTable must be detected by the audit and + * reported as an inconsistency, with the previewFailures metric bumped on the + * coordinator and diagnostic snapshots taken on participating replicas. + *

+ * TODO: injectSSTableCellDivergence is stubbed and throws + * UnsupportedOperationException; the test will fail at the injection call + * until the primitive is implemented. Documents the missing capability. + */ + @Test + public void previewDetectsDivergenceInReconciledSSTable() + { + createTrackedSchema(CLUSTER, keyspace, table); + + for (int i = 0; i < 10; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + settleReconciliation(CLUSTER); + + // Force everything to disk so the diverged data lives in an SSTable, not the journal. + flushAll(CLUSTER, keyspace, table); + + // Corrupt an already-reconciled cell on one replica so the audit sees a mismatch + // in the SSTable stream (as distinct from the journal stream). + injectSSTableCellDivergence(CLUSTER.get(2), keyspace, table, 3, 999); + + long[] previewFailuresBefore = previewFailuresPerNode(CLUSTER); + + NodeToolResult result = CLUSTER.get(1).nodetoolResult(true, "repair", "--validate", keyspace); + result.asserts().success(); + + // Mismatch messaging (reused wording from classic PreviewRepairTask). + assertNotificationContains(result, "Repaired data is inconsistent"); + + // Coordinator increments the metric; other nodes should not. + assertPreviewFailuresBumpedOnCoordinator(CLUSTER, previewFailuresBefore, 1); + + // Diagnostic snapshots on every participating replica. + assertDiagnosticSnapshotExists(CLUSTER, keyspace, table, 1, 2, 3); + } + + /** + * Divergence in an already-reconciled journal entry must be detected by the + * audit and reported as an inconsistency. Unlike the SSTable-side variant, + * this test skips flushAll so the reconciled mutation stays in the journal + * and feeds the audit's journal stream rather than the SSTable stream. + *

+ * TODO: injectJournalPayloadDivergence is stubbed and throws + * UnsupportedOperationException; the test will fail at the injection call + * until the primitive is implemented. + */ + @Test + public void previewDetectsDivergenceInReconciledJournalEntry() + { + createTrackedSchema(CLUSTER, keyspace, table); + + for (int i = 0; i < 10; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + settleReconciliation(CLUSTER); + + // Deliberately no flush — reconciled mutations must stay in the journal so + // the injection can rewrite an existing journal entry's payload rather than + // touching an SSTable. + injectJournalPayloadDivergence(CLUSTER.get(2), keyspace, table, 3, 999); + + long[] previewFailuresBefore = previewFailuresPerNode(CLUSTER); + + NodeToolResult result = CLUSTER.get(1).nodetoolResult(true, "repair", "--validate", keyspace); + result.asserts().success(); + + assertNotificationContains(result, "Repaired data is inconsistent"); + assertPreviewFailuresBumpedOnCoordinator(CLUSTER, previewFailuresBefore, 1); + assertDiagnosticSnapshotExists(CLUSTER, keyspace, table, 1, 2, 3); + } + + /** + * SSTables flushed after cut establishment must not be part of the audit, + * because the audit's SSTable set is snapshotted (via Refs) on each participant + * at IR start. This test forces node 2 to flush a new SSTable after the cut is + * established, corrupts a cell in that SSTable, and asserts the audit reports + * in-sync — the new SSTable is not in the snapshot, so the corruption is + * invisible to the merkle tree. + *

+ * TODO: injectSSTableCellDivergence is stubbed and throws + * UnsupportedOperationException; the test will fail at the injection call + * until the primitive is implemented. + */ + @Test + public void previewIgnoresDivergenceInSSTablesFlushedAfterCutEstablishment() throws Exception + { + createTrackedSchema(CLUSTER, keyspace, table); + + // Initial reconciled writes; deliberately no flush so this run's SSTables + // all arise later. The point of the test is that any SSTable born after + // cut establishment on any participant is excluded from the audit. + for (int i = 0; i < 10; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + settleReconciliation(CLUSTER); + + long[] previewFailuresBefore = previewFailuresPerNode(CLUSTER); + + IInvokableInstance coordinator = CLUSTER.get(1); + coordinator.runOnInstance(CutEstablishedLatch::install); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + + try + { + Future resultFuture = executor.submit(() -> coordinator.nodetoolResult(true, "repair", "--validate", keyspace)); + + // Block until "cut established" — by this notification, each participant + // has already captured its SSTable-set-via-Refs and journal snapshot as + // part of MT_PREVIEW_PREPARE_REQ handling, before IR ran. + coordinator.runOnInstance(CutEstablishedLatch::await); + + // Post-cut writes at CL.ALL: offsets > cut, land in every replica's memtable. + for (int i = 10; i < 20; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + // Flush only on node 2. The resulting SSTable is brand new — created after + // node 2's snapshot was taken — so it is not in the audit's Refs set and + // must not contribute to the merkle tree. + CLUSTER.get(2).nodetoolResult("flush", keyspace, table).asserts().success(); + + // Corrupt a cell in node 2's post-snapshot SSTable. If the audit correctly + // reads only its snapshotted SSTable set, this divergence is invisible. + injectSSTableCellDivergence(CLUSTER.get(2), keyspace, table, 15, 999); + + NodeToolResult result = resultFuture.get(120, TimeUnit.SECONDS); + result.asserts().success(); + + assertNotificationContains(result, "cut established"); + assertNotificationContains(result, "audit complete"); + assertNotificationContains(result, "Repaired data is in sync"); + + assertPreviewFailuresUnchanged(CLUSTER, previewFailuresBefore); + assertNoDiagnosticSnapshot(CLUSTER, keyspace, table); + } + finally + { + coordinator.runOnInstance(CutEstablishedLatch::uninstall); + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + /** + * Journal entries written after cut establishment must not be part of the audit, + * because the audit's journal source is a MutationJournal.Snapshot pinned on each + * participant at IR start (via MT_PREVIEW_PREPARE_REQ). This test drives post-cut + * writes so new mutations land in every replica's current journal, then corrupts + * one such mutation on node 2. The audit should still report in-sync because the + * pinned journal snapshot on node 2 doesn't contain the post-cut entry. + *

+ * TODO: injectJournalPayloadDivergence is stubbed and throws + * UnsupportedOperationException; the test will fail at the injection call + * until the primitive is implemented. + */ + @Test + public void previewIgnoresDivergenceInJournalEntriesWrittenAfterCutEstablishment() throws Exception + { + createTrackedSchema(CLUSTER, keyspace, table); + + for (int i = 0; i < 10; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + settleReconciliation(CLUSTER); + + long[] previewFailuresBefore = previewFailuresPerNode(CLUSTER); + + IInvokableInstance coordinator = CLUSTER.get(1); + coordinator.runOnInstance(CutEstablishedLatch::install); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + + try + { + Future resultFuture = executor.submit(() -> coordinator.nodetoolResult(true, "repair", "--validate", keyspace)); + + // Block until "cut established" — by this notification, each participant + // has pinned its MutationJournal.Snapshot as part of MT_PREVIEW_PREPARE_REQ + // handling, before IR ran. + coordinator.runOnInstance(CutEstablishedLatch::await); + + // Post-cut writes at CL.ALL: new mutations added to every replica's current + // journal at offsets > cut, but NOT in any participant's pinned snapshot. + for (int i = 10; i < 20; i++) + CLUSTER.coordinator(1).execute("INSERT INTO " + keyspace + '.' + table + " (k, v) VALUES (?, ?)", ConsistencyLevel.ALL, i, i); + + // No flush — the post-cut mutations stay in the current journal. + // Corrupt one of them on node 2. Because the audit reads from node 2's + // pinned snapshot (taken before this mutation existed), the corruption + // is invisible. + injectJournalPayloadDivergence(CLUSTER.get(2), keyspace, table, 15, 999); + + NodeToolResult result = resultFuture.get(120, TimeUnit.SECONDS); + result.asserts().success(); + + assertNotificationContains(result, "cut established"); + assertNotificationContains(result, "audit complete"); + assertNotificationContains(result, "Repaired data is in sync"); + + assertPreviewFailuresUnchanged(CLUSTER, previewFailuresBefore); + assertNoDiagnosticSnapshot(CLUSTER, keyspace, table); + } + finally + { + coordinator.runOnInstance(CutEstablishedLatch::uninstall); + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + /** + * A tracked keyspace that is currently undergoing mutation-tracking migration + * (mutationTrackingMigrationState.isMigrating(keyspace) == true) must have + * --validate rejected up front — before any cut work is attempted. The keyspace + * is put into MIGRATING_TO_TRACKED state via ALTER KEYSPACE and left there + * (no incremental repair runs to advance the migration). + */ + @Test + public void previewRejectsWhenKeyspaceIsMigrating() + { + // Create untracked first, then ALTER to tracked → keyspace enters migration. + CLUSTER.schemaChange("CREATE KEYSPACE " + keyspace + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + NODES + "} AND replication_type='untracked'"); + CLUSTER.schemaChange("CREATE TABLE " + keyspace + '.' + table + " (k int PRIMARY KEY, v int)"); + CLUSTER.schemaChange("ALTER KEYSPACE " + keyspace + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + NODES + "} AND replication_type='tracked'"); + + // Sanity: the keyspace really is in migration state on the coordinator. + String ks = keyspace; + boolean isMigrating = CLUSTER.get(1).callOnInstance(() -> ClusterMetadata.current().mutationTrackingMigrationState.isMigrating(ks)); + assertTrue("Expected keyspace " + keyspace + " to be in mutation tracking migration state", isMigrating); + + NodeToolResult result = CLUSTER.get(1).nodetoolResult(true, "repair", "--validate", keyspace); + result.asserts().failure().errorContains("migration"); + + // Rejection must fire in the gate step — before any cut work is attempted, + // so the MT preview task's earliest progress notification is never emitted. + assertNotificationAbsent(result, "establishing cut"); + } + + private static void createTrackedSchema(Cluster cluster, String keyspace, String table) + { + cluster.schemaChange("CREATE KEYSPACE " + keyspace + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': " + NODES + "} AND replication_type='tracked'"); + cluster.schemaChange("CREATE TABLE " + keyspace + '.' + table + " (k int PRIMARY KEY, v int)"); + } + + /** + * Force outstanding tracked writes through the background-reconciliation path so + * that every replica's reconciled offsets catch up before the next assertion. + * Mirrors the pattern in TrackedTransferTestBase.assertCompaction. + */ + private static void settleReconciliation(Cluster cluster) + { + cluster.forEach(i -> i.runOnInstance(() ->{ + MutationTrackingService.instance().persistLogStateForTesting(); + MutationTrackingService.instance().broadcastOffsetsForTesting(); + })); + + // TODO: Do this without sleeping? + Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS); + } + + /** + * Force any outstanding memtables for (keyspace, table) to disk on every replica + * so subsequent test steps operate against SSTables rather than the journal. + */ + private static void flushAll(Cluster cluster, String keyspace, String table) + { + cluster.forEach(i -> i.nodetoolResult("flush", keyspace, table).asserts().success()); + } + + /** + * Stub: modify this instance's on-disk SSTable for (keyspace, table) so the cell + * for key=key reads divergentValue instead of the reconciled value, without + * disturbing the SSTable's coordinatorLogOffsets metadata (so it stays "fully + * below cut" from the audit's perspective). + *

+ * TODO: not yet implemented — throws until the production hook is added. + */ + private static void injectSSTableCellDivergence(IInvokableInstance instance, String keyspace, String table, int key, int divergentValue) + { + throw new UnsupportedOperationException( + "SSTable cell divergence injection not yet implemented. Should modify this instance's on-disk SSTable for (" + + keyspace + '.' + table + ") so the cell for key=" + key + " reads " + divergentValue + " instead of the reconciled value."); + } + + /** + * Stub: rewrite the payload of an existing reconciled mutation in this + * instance's journal for (keyspace, table) key=key so it reads divergentValue + * instead of the reconciled value. Must NOT change the mutation's offset or + * alter any offset bookkeeping — the offset must still appear reconciled to + * every replica, so the cut includes it and the audit reads the divergent + * payload. + *

+ * TODO: not yet implemented — throws until the production hook is added. + */ + private static void injectJournalPayloadDivergence(IInvokableInstance instance, String keyspace, String table, int key, int divergentValue) + { + throw new UnsupportedOperationException( + "Journal payload divergence injection not yet implemented. Should rewrite the payload of an existing reconciled mutation in this instance's journal for (" + + keyspace + '.' + table + ") key=" + key + " so it reads " + divergentValue + " instead of the reconciled value, without changing the offset or altering any offset bookkeeping."); + } + + private static void assertNotificationContains(NodeToolResult result, String expectedFragment) + { + List notifications = result.getNotifications(); + for (Notification n : notifications) + { + String message = n.getMessage(); + if (message != null && message.contains(expectedFragment)) + return; + } + + StringBuilder found = new StringBuilder(); + for (Notification n : notifications) + { + if (n.getMessage() != null) + found.append("\n - ").append(n.getMessage()); + } + fail("Expected notification containing \"" + expectedFragment + "\" but found:" + found); + } + + private static void assertNotificationAbsent(NodeToolResult result, String unexpectedFragment) + { + for (Notification n : result.getNotifications()) + { + String message = n.getMessage(); + if (message != null && message.contains(unexpectedFragment)) + fail("Did not expect notification containing \"" + unexpectedFragment + "\" but found: " + message); + } + } + + @SuppressWarnings("Convert2MethodRef") + private static long[] previewFailuresPerNode(Cluster cluster) + { + long[] counts = new long[cluster.size()]; + for (int i = 0; i < cluster.size(); i++) + { + counts[i] = cluster.get(i + 1).callOnInstance(() -> RepairMetrics.previewFailures.getCount()); + } + return counts; + } + + private static void assertPreviewFailuresUnchanged(Cluster cluster, long[] baseline) + { + long[] after = previewFailuresPerNode(cluster); + for (int i = 0; i < after.length; i++) + assertThat(after[i]).as("RepairMetrics.previewFailures on node %d", i + 1).isEqualTo(baseline[i]); + } + + /** + * Assert that RepairMetrics.previewFailures incremented by exactly 1 on the + * coordinator node and is unchanged on every other node in the cluster. + */ + private static void assertPreviewFailuresBumpedOnCoordinator(Cluster cluster, long[] baseline, int coordinatorNode) + { + long[] after = previewFailuresPerNode(cluster); + for (int i = 0; i < after.length; i++) + { + long expectedDelta = (i + 1 == coordinatorNode) ? 1 : 0; + assertThat(after[i]).as("RepairMetrics.previewFailures on node %d", i + 1).isEqualTo(baseline[i] + expectedDelta); + } + } + + private static void assertNoDiagnosticSnapshot(Cluster cluster, String keyspace, String table) + { + for (IInvokableInstance instance : cluster) + { + instance.runOnInstance(() -> { + List matching = SnapshotManager.instance.getSnapshots(snapshot -> + snapshot.getKeyspaceName().equals(keyspace) + && snapshot.getTableName().equals(table) + && snapshot.getTag().startsWith(DiagnosticSnapshotService.REPAIRED_DATA_MISMATCH_SNAPSHOT_PREFIX)); + + if (!matching.isEmpty()) + { + StringJoiner tags = new StringJoiner(", "); + for (TableSnapshot s : matching) + tags.add(s.getTag()); + throw new AssertionError("Unexpected diagnostic snapshot(s) on this replica: " + tags); + } + }); + } + } + + /** + * Assert that a REPAIRED_DATA_MISMATCH diagnostic snapshot exists for + * (keyspace, table) on every node listed in expectedNodes. + */ + private static void assertDiagnosticSnapshotExists(Cluster cluster, String keyspace, String table, int... expectedNodes) + { + for (int nodeNum : expectedNodes) + { + cluster.get(nodeNum).runOnInstance(() -> { + List matching = SnapshotManager.instance.getSnapshots(snapshot -> + snapshot.getKeyspaceName().equals(keyspace) + && snapshot.getTableName().equals(table) + && snapshot.getTag().startsWith(DiagnosticSnapshotService.REPAIRED_DATA_MISMATCH_SNAPSHOT_PREFIX)); + + if (matching.isEmpty()) + throw new AssertionError("Expected REPAIRED_DATA_MISMATCH diagnostic snapshot for " + keyspace + '.' + table + " on this replica, but found none."); + }); + } + } + + /** + * Instance-side latch that fires when StorageService emits a JMX notification + * whose message contains "cut established" — the milestone + * MutationTrackingRepairedPreviewTask reports when the audit cut is fixed. + *

+ * Static state lives in the instance's classloader; install/await/uninstall + * must all be invoked via runOnInstance against the same instance. + */ + public static class CutEstablishedLatch + { + private static volatile CountDownLatch latch; + private static volatile NotificationListener listener; + + public static void install() + { + latch = new CountDownLatch(1); + listener = (notification, handback) -> { + String msg = notification.getMessage(); + if (msg != null && msg.contains("cut established")) + latch.countDown(); + }; + StorageService.instance.addNotificationListener(listener, null, null); + } + + public static void await() + { + try + { + if (!latch.await(60, TimeUnit.SECONDS)) + throw new AssertionError("Timed out waiting for 'cut established' notification"); + } + catch (InterruptedException e) + { + throw new AssertionError("Interrupted while waiting for 'cut established' notification"); + } + } + + public static void uninstall() + { + try + { + if (listener != null) + StorageService.instance.removeNotificationListener(listener); + } + catch (Exception ignore) + { + } + listener = null; + latch = null; + } + } +}