Bring back the ability to Optionally avoid hint transfer during decommission (CASSANDRA-17808)

lost due to merge of ae08423

patch by Matt Byrd; reviewed by Stefan Miklosovic and Sam Tunnicliffe for CASSANDRA-21341

Author: Matt Byrd

CHANGES.txt

@@ -1,4 +1,5 @@
 6.0-alpha2
+ * Restore option to avoid hint transfer during decommission (CASSANDRA-21341)
  * Add an offline cluster metadata tool (CASSANDRA-19151)
  * Accord: Tail Latency Improvements (CASSANDRA-21361)
  * Artificial Latency Injection (CASSANDRA-17024)

src/java/org/apache/cassandra/tcm/sequences/UnbootstrapStreams.java

@@ -30,9 +30,11 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.batchlog.BatchlogManager;
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.hints.HintsService;
 import org.apache.cassandra.locator.EndpointsByReplica;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.RangesAtEndpoint;
@@ -50,6 +52,7 @@
 import org.apache.cassandra.tcm.ownership.MovementMap;
 import org.apache.cassandra.tcm.ownership.PlacementDeltas;
 import org.apache.cassandra.utils.concurrent.Future;
+import org.apache.cassandra.utils.concurrent.ImmediateFuture;
 
 public class UnbootstrapStreams implements LeaveStreams
 {
@@ -125,9 +128,20 @@ private static void unbootstrap(Keyspaces keyspaces, MovementMap movements) thro
         logger.debug("waiting for batch log processing.");
         batchlogReplay.get();
 
-        logger.info("streaming hints to other nodes");
+        Future<?> hintsSuccess = ImmediateFuture.success(null);
 
-        Future<?> hintsSuccess = StorageService.instance.streamHints();
+        if (DatabaseDescriptor.getTransferHintsOnDecommission())
+        {
+            logger.info("streaming hints to other nodes");
+            hintsSuccess = StorageService.instance.streamHints();
+        }
+        else
+        {
+            logger.info("pausing dispatch and deleting hints");
+            DatabaseDescriptor.setHintedHandoffEnabled(false);
+            HintsService.instance.pauseDispatch();
+            HintsService.instance.deleteAllHints();
+        }
 
         // wait for the transfer runnables to signal the latch.
         logger.debug("waiting for stream acks.");

test/distributed/org/apache/cassandra/distributed/test/HintedHandoffAddRemoveNodesTest.java

@@ -71,25 +71,27 @@ public void shouldAvoidHintTransferOnDecommission() throws Exception
         {
             cluster.schemaChange(withKeyspace("CREATE TABLE %s.decom_no_hints_test (key int PRIMARY KEY, value int)"));
 
-            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.decom_no_hints_test (key, value) VALUES (?, ?)"), ALL, 0, 0);
-            long hintsBeforeShutdown = countTotalHints(cluster.get(1));
+            int secondNode = 2;
+            cluster.coordinator(secondNode).execute(withKeyspace("INSERT INTO %s.decom_no_hints_test (key, value) VALUES (?, ?)"), ALL, 0, 0);
+            long hintsBeforeShutdown = countTotalHints(cluster.get(secondNode));
             assertThat(hintsBeforeShutdown).isEqualTo(0);
-            long hintsDelivered = countHintsDelivered(cluster.get(1));
+            long hintsDelivered = countHintsDelivered(cluster.get(secondNode));
             assertThat(hintsDelivered).isEqualTo(0);
 
             // Shutdown node 3 so hints can be written against it.
             cluster.get(3).shutdown().get();
 
-            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.decom_no_hints_test (key, value) VALUES (?, ?)"), TWO, 0, 0);
-            Awaitility.await().until(() -> countTotalHints(cluster.get(1)) > 0);
-            long hintsAfterShutdown = countTotalHints(cluster.get(1));
+            cluster.coordinator(secondNode).execute(withKeyspace("INSERT INTO %s.decom_no_hints_test (key, value) VALUES (?, ?)"), TWO, 0, 0);
+            Awaitility.await().until(() -> countTotalHints(cluster.get(secondNode)) > 0);
+            long hintsAfterShutdown = countTotalHints(cluster.get(secondNode));
             assertThat(hintsAfterShutdown).isEqualTo(1);
 
-            cluster.get(2).runOnInstance(() -> setProgressBarrierMinConsistencyLevel(org.apache.cassandra.db.ConsistencyLevel.ONE));
+            cluster.get(secondNode).runOnInstance(() -> setProgressBarrierMinConsistencyLevel(org.apache.cassandra.db.ConsistencyLevel.ONE));
+
             ClusterUtils.waitForCMSToQuiesce(cluster, cluster.get(1), 3);
-            cluster.get(2).nodetoolResult("decommission", "--force").asserts().success();
-            long hintsDeliveredByDecom = countHintsDelivered(cluster.get(2));
-            String mode = cluster.get(2).callOnInstance(() -> StorageService.instance.getOperationMode());
+            cluster.get(secondNode).nodetoolResult("decommission", "--force").asserts().success();
+            long hintsDeliveredByDecom = countHintsDelivered(cluster.get(secondNode));
+            String mode = cluster.get(secondNode).callOnInstance(() -> StorageService.instance.getOperationMode());
             assertEquals(StorageService.Mode.DECOMMISSIONED.toString(), mode);
             assertThat(hintsDeliveredByDecom).isEqualTo(0);
         }