C21563 misc cep45 bugfixes

src/java/org/apache/cassandra/replication/MutationTrackingService.java

@@ -462,7 +462,12 @@ public void updateReplicatedOffsets(String keyspace, Range<Token> range, List<?
         shardLock.readLock().lock();
         try
         {
-            getOrCreateShards(keyspace).updateReplicatedOffsets(range, offsets, durable, onHost);
+            KeyspaceShards shards = getOrCreateShards(keyspace);
+            // The keyspace may have been dropped locally while this broadcast was in flight; ignore it.
+            if (shards == null)
+                logger.debug("Ignoring replicated offsets broadcast from {} for unknown (likely dropped) keyspace {}", onHost, keyspace);
+            else
+                shards.updateReplicatedOffsets(range, offsets, durable, onHost);
         }
         finally
         {
@@ -830,14 +835,19 @@ private KeyspaceShards getOrCreateShards(TableId tableId)
         return getOrCreateShards(Schema.instance.getTableMetadata(tableId).keyspace);
     }
 
+    @Nullable
     private KeyspaceShards getOrCreateShards(String keyspace)
     {
         KeyspaceShards ks = keyspaceShards.get(keyspace);
         if (ks != null)
             return ks;
 
         ClusterMetadata csm = ClusterMetadata.current();
-        KeyspaceMetadata ksm = csm.schema.getKeyspaceMetadata(keyspace);
+        // The keyspace may have been dropped locally while messages referencing it (e.g. offset broadcasts
+        // from peers) are still in flight. Return null so inbound handlers can ignore them gracefully.
+        KeyspaceMetadata ksm = csm.schema.maybeGetKeyspaceMetadata(keyspace).orElse(null);
+        if (ksm == null)
+            return null;
         return keyspaceShards.computeIfAbsent(keyspace, ignore -> KeyspaceShards.make(ksm, csm, this::nextLogId, this::onNewLog));
     }
 

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()

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);

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<Token> fullTokenRange()
     {
         return new Range<>(
@@ -233,17 +228,17 @@ 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(
                     "INSERT INTO " + tableName("4") + " (k, v) VALUES (?, ?)",
                     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<Token> range = fullTokenRange();
                 RepairJobDesc desc = new RepairJobDesc(TimeUUID.Generator.nextTimeUUID(),

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<Object[]> data()
     {
         List<Object[]> result = new ArrayList<>();

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";
     }

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);

test/resources/nodetool/help/mtadmin

@@ -0,0 +1,43 @@
+NAME
+        nodetool mtadmin - Manage mutation tracking
+
+SYNOPSIS
+        nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]
+                [(-pw <password> | --password <password>)]
+                [(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
+                [(-u <username> | --username <username>)] mtadmin <command> [<args>]
+
+        nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]
+                [(-pw <password> | --password <password>)]
+                [(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
+                [(-u <username> | --username <username>)] mtadmin getconfig
+
+        nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]
+                [(-pw <password> | --password <password>)]
+                [(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
+                [(-u <username> | --username <username>)] mtadmin setconfig [--]
+                <paramType> <paramValue>
+
+OPTIONS
+        -h <host>, --host <host>
+            Node hostname or ip address
+
+        -p <port>, --port <port>
+            Remote jmx agent port number
+
+        -pw <password>, --password <password>
+            Remote jmx agent password
+
+        -pwf <passwordFilePath>, --password-file <passwordFilePath>
+            Path to the JMX password file
+
+        -u <username>, --username <username>
+            Remote jmx agent username
+
+COMMANDS
+        With no arguments, Display help information
+
+        getconfig
+            Print mutation tracking configurations
+        setconfig
+            Sets the mutation tracking configuration

test/resources/nodetool/help/mtadmin$getconfig

@@ -0,0 +1,24 @@
+NAME
+        nodetool mtadmin getconfig - Print mutation tracking configurations
+
+SYNOPSIS
+        nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]
+                [(-pw <password> | --password <password>)]
+                [(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
+                [(-u <username> | --username <username>)] mtadmin getconfig
+
+OPTIONS
+        -h <host>, --host <host>
+            Node hostname or ip address
+
+        -p <port>, --port <port>
+            Remote jmx agent port number
+
+        -pw <password>, --password <password>
+            Remote jmx agent password
+
+        -pwf <passwordFilePath>, --password-file <passwordFilePath>
+            Path to the JMX password file
+
+        -u <username>, --username <username>
+            Remote jmx agent username

test/resources/nodetool/help/mtadmin$setconfig

@@ -0,0 +1,36 @@
+NAME
+        nodetool mtadmin setconfig - Sets the mutation tracking configuration
+
+SYNOPSIS
+        nodetool [(-h <host> | --host <host>)] [(-p <port> | --port <port>)]
+                [(-pw <password> | --password <password>)]
+                [(-pwf <passwordFilePath> | --password-file <passwordFilePath>)]
+                [(-u <username> | --username <username>)] mtadmin setconfig [--]
+                <paramType> <paramValue>
+
+OPTIONS
+        -h <host>, --host <host>
+            Node hostname or ip address
+
+        -p <port>, --port <port>
+            Remote jmx agent port number
+
+        -pw <password>, --password <password>
+            Remote jmx agent password
+
+        -pwf <passwordFilePath>, --password-file <passwordFilePath>
+            Path to the JMX password file
+
+        -u <username>, --username <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
+
+        <paramType>
+            Mutation tracking param type.
+
+        <paramValue>
+            Mutation tracking param value

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

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