diff --git a/src/java/org/apache/cassandra/replication/MutationTrackingService.java b/src/java/org/apache/cassandra/replication/MutationTrackingService.java index db87a88f010a..010fe31dcb22 100644 --- a/src/java/org/apache/cassandra/replication/MutationTrackingService.java +++ b/src/java/org/apache/cassandra/replication/MutationTrackingService.java @@ -462,7 +462,12 @@ public void updateReplicatedOffsets(String keyspace, Range range, List KeyspaceShards.make(ksm, csm, this::nextLogId, this::onNewLog)); } diff --git a/src/java/org/apache/cassandra/replication/ShortMutationId.java b/src/java/org/apache/cassandra/replication/ShortMutationId.java index 856790a8a1df..c35ad84978b4 100644 --- a/src/java/org/apache/cassandra/replication/ShortMutationId.java +++ b/src/java/org/apache/cassandra/replication/ShortMutationId.java @@ -76,7 +76,7 @@ private ShortMutationId(int hostId, int hostLogId, int offset) public ShortMutationId(MutationId mutationId) { - this(mutationId.hostLogId(), mutationId.hostId(), mutationId.offset()); + this(mutationId.hostId(), mutationId.hostLogId(), mutationId.offset()); } public int hostId() diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java index f5ad9f7d26ff..87b1eb0457be 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java @@ -881,11 +881,12 @@ public void testRepairSyncTimeout() @Test public void testRepairFailsOnTopologyChange() throws Exception { - insertData("tbl", 0, 50); - // Block offset broadcasts so the sync coordinator stays alive waiting IMessageFilters.Filter offsetFilter = CLUSTER.filters().verbs(Verb.MT_BROADCAST_LOG_OFFSETS.id).drop(); + // must use inconsistent data to prevent write process from marking the writes reconciled + insertDataWithInconsistency("tbl", 0, 50); + // Use a latch to detect when the sync request has been sent, meaning // the sync coordinator is active and tracking shard references CountDownLatch syncStarted = new CountDownLatch(1); diff --git a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java index 7a13f606f2b1..c1d82ef104fc 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/replication/MutationTrackingSyncCoordinatorTest.java @@ -29,6 +29,7 @@ import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.net.Verb; import org.apache.cassandra.repair.RepairJobDesc; import org.apache.cassandra.repair.SharedContext; import org.apache.cassandra.replication.MutationTrackingService; @@ -64,12 +65,6 @@ private String tableName(String suffix) return KS_NAME + suffix + '.' + TBL_NAME; } - private void pauseOffsetBroadcasts(Cluster cluster, boolean pause) - { - for (int i = 1; i <= cluster.size(); i++) - cluster.get(i).runOnInstance(() -> MutationTrackingService.instance().pauseOffsetBroadcast(pause)); - } - private static Range fullTokenRange() { return new Range<>( @@ -233,9 +228,6 @@ public void testSyncCoordinatorCancel() throws Throwable { createTrackedKeyspace(cluster, "4"); - // Pause offset broadcasts on all nodes to prevent sync from completing - pauseOffsetBroadcasts(cluster, true); - for (int i = 0; i < 100; i++) { cluster.coordinator(1).execute( @@ -243,7 +235,10 @@ public void testSyncCoordinatorCancel() throws Throwable ConsistencyLevel.ONE, i, i); } - // Start coordinator - it will be stuck waiting for offsets + // Drop sync responses so the coordinator never receives them + cluster.filters().verbs(Verb.MT_SYNC_RSP.id).drop(); + + // Start coordinator - it will be stuck waiting for sync responses Boolean wasCancelled = cluster.get(1).callOnInstance(() -> { Range range = fullTokenRange(); RepairJobDesc desc = new RepairJobDesc(TimeUUID.Generator.nextTimeUUID(), diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java index db3d841bddf2..ebec55383b06 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/IndexStreamingTest.java @@ -69,7 +69,7 @@ private static int sstableStreamingComponentsCount() @Parameterized.Parameter(2) public boolean isWide; - @Parameterized.Parameters(name = "isLiteral={0}, isZeroCopyStreaming={1}") + @Parameterized.Parameters(name = "isLiteral={0}, isZeroCopyStreaming={1}, isWide={2}") public static List data() { List result = new ArrayList<>(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java index f3d6adc1215c..b9c89b7778d1 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingUtils.java @@ -60,6 +60,7 @@ public class MutationTrackingUtils public static class IgnoreReasons { + public static final String NO_TOPOLOGY_CHANGES = "NO_TOPOLOGY_CHANGES"; public static final String NO_RANGE_MOVEMENTS = "NO_RANGE_MOVEMENTS"; public static final String NO_PER_PARTITION_RANGE_READ_LIMITS = "NO_PER_PARTITION_RANGE_READ_LIMITS"; } diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java index ee897a31886a..4c1f29ff3c27 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java @@ -79,7 +79,7 @@ public void importMissedActivationCommit() throws Throwable public void importMissedActivation(ActivationRequest.Phase phase) throws Throwable { int MISSED_ACTIVATION = 2; - try (Cluster cluster = cluster(TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION))) + try (Cluster cluster = disableBackgroundReconciler(cluster(TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.install(MISSED_ACTIVATION)))) { TrackedTransferTestBase.ByteBuddyInjections.SkipActivation.setup(cluster, phase); createSchema(cluster); diff --git a/test/resources/nodetool/help/mtadmin b/test/resources/nodetool/help/mtadmin new file mode 100644 index 000000000000..0d2b141ea7e6 --- /dev/null +++ b/test/resources/nodetool/help/mtadmin @@ -0,0 +1,43 @@ +NAME + nodetool mtadmin - Manage mutation tracking + +SYNOPSIS + nodetool [(-h | --host )] [(-p | --port )] + [(-pw | --password )] + [(-pwf | --password-file )] + [(-u | --username )] mtadmin [] + + nodetool [(-h | --host )] [(-p | --port )] + [(-pw | --password )] + [(-pwf | --password-file )] + [(-u | --username )] mtadmin getconfig + + nodetool [(-h | --host )] [(-p | --port )] + [(-pw | --password )] + [(-pwf | --password-file )] + [(-u | --username )] mtadmin setconfig [--] + + +OPTIONS + -h , --host + Node hostname or ip address + + -p , --port + Remote jmx agent port number + + -pw , --password + Remote jmx agent password + + -pwf , --password-file + Path to the JMX password file + + -u , --username + Remote jmx agent username + +COMMANDS + With no arguments, Display help information + + getconfig + Print mutation tracking configurations + setconfig + Sets the mutation tracking configuration diff --git a/test/resources/nodetool/help/mtadmin$getconfig b/test/resources/nodetool/help/mtadmin$getconfig new file mode 100644 index 000000000000..46d743bfba52 --- /dev/null +++ b/test/resources/nodetool/help/mtadmin$getconfig @@ -0,0 +1,24 @@ +NAME + nodetool mtadmin getconfig - Print mutation tracking configurations + +SYNOPSIS + nodetool [(-h | --host )] [(-p | --port )] + [(-pw | --password )] + [(-pwf | --password-file )] + [(-u | --username )] mtadmin getconfig + +OPTIONS + -h , --host + Node hostname or ip address + + -p , --port + Remote jmx agent port number + + -pw , --password + Remote jmx agent password + + -pwf , --password-file + Path to the JMX password file + + -u , --username + Remote jmx agent username diff --git a/test/resources/nodetool/help/mtadmin$setconfig b/test/resources/nodetool/help/mtadmin$setconfig new file mode 100644 index 000000000000..ea30d9c77941 --- /dev/null +++ b/test/resources/nodetool/help/mtadmin$setconfig @@ -0,0 +1,36 @@ +NAME + nodetool mtadmin setconfig - Sets the mutation tracking configuration + +SYNOPSIS + nodetool [(-h | --host )] [(-p | --port )] + [(-pw | --password )] + [(-pwf | --password-file )] + [(-u | --username )] mtadmin setconfig [--] + + +OPTIONS + -h , --host + Node hostname or ip address + + -p , --port + Remote jmx agent port number + + -pw , --password + Remote jmx agent password + + -pwf , --password-file + Path to the JMX password file + + -u , --username + Remote jmx agent username + + -- + This option can be used to separate command-line options from the + list of argument, (useful when arguments might be mistaken for + command-line options + + + Mutation tracking param type. + + + Mutation tracking param value diff --git a/test/resources/nodetool/help/nodetool b/test/resources/nodetool/help/nodetool index a3fe2c520e28..023cc4f82fe4 100644 --- a/test/resources/nodetool/help/nodetool +++ b/test/resources/nodetool/help/nodetool @@ -95,6 +95,7 @@ The most commonly used nodetool commands are: listpendinghints Print all pending hints that this node has listsnapshots Lists all the snapshots along with the size on disk and true size. True size is the total size of all SSTables which are not backed up to disk. Size on disk is total size of the snapshot on disk. Total TrueDiskSpaceUsed does not make any SSTable deduplication. move Move node on the token ring to a new token + mtadmin Manage mutation tracking netstats Print network information on provided host (connecting node by default) pausehandoff Pause hints delivery process profile Manage Async-Profiler on a Cassandra process diff --git a/test/unit/org/apache/cassandra/ServerTestUtils.java b/test/unit/org/apache/cassandra/ServerTestUtils.java index 448ac0468e77..5d1d8e8fef68 100644 --- a/test/unit/org/apache/cassandra/ServerTestUtils.java +++ b/test/unit/org/apache/cassandra/ServerTestUtils.java @@ -155,9 +155,6 @@ public static void prepareServerNoRegister() { daemonInitialization(); - // Need to happen after daemonInitialization for config to be set, but before CFS initialization - MutationJournal.start(); - if (isServerPrepared) return; @@ -196,6 +193,8 @@ public void uncaughtException(Thread t, Throwable e) ThreadAwareSecurityManager.install(); CassandraRelevantProperties.GOSSIPER_SKIP_WAITING_TO_SETTLE.setInt(0); + // Need to happen after daemonInitialization for config to be set, but before CFS initialization + MutationJournal.start(); initCMS(); SystemKeyspace.persistLocalMetadata(); AuditLogManager.instance.initialize(); @@ -230,6 +229,7 @@ public static void cleanup() cleanupDirectory(cdcDir); cleanupDirectory(DatabaseDescriptor.getHintsDirectory()); cleanupDirectory(DatabaseDescriptor.getAccordJournalDirectory()); + cleanupDirectory(DatabaseDescriptor.getMutationTrackingJournalDirectory()); cleanupSavedCaches(); // clean up data directory which are stored as data directory/keyspace/data files