From 4164313f926b605a24395a8ca920e8bea0691204 Mon Sep 17 00:00:00 2001
From: Brandon Williams <brandonwilliams@apache.org>
Date: Fri, 9 Apr 2021 16:19:17 -0500
Subject: [PATCH] Don't wait for migrations from removed nodes, mention flag to
 skip

Patch by brandonwilliams; reviewed by Adam Holmberg, adelapena and
bdeggleston for CASSANDRA-16577
---
 CHANGES.txt                                   |  1 +
 .../service/MigrationCoordinator.java         |  9 ++++++++
 .../cassandra/service/StorageService.java     |  8 ++++---
 .../service/MigrationCoordinatorTest.java     | 21 +++++++++++++++++++
 4 files changed, 36 insertions(+), 3 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index f7ad917743..12fd27a49a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.0.25:
+ * Don't wait for schema migrations from removed nodes (CASSANDRA-16577)
  * Scheduled (delayed) schema pull tasks should not run after MIGRATION stage shutdown during decommission (CASSANDRA-16495)
  * Ignore trailing zeros in hint files (CASSANDRA-16523)
  * Refuse DROP COMPACT STORAGE if some 2.x sstables are in use (CASSANDRA-15897)
diff --git a/src/java/org/apache/cassandra/service/MigrationCoordinator.java b/src/java/org/apache/cassandra/service/MigrationCoordinator.java
index f3f5cdfc2d..94c90ae5dd 100644
--- a/src/java/org/apache/cassandra/service/MigrationCoordinator.java
+++ b/src/java/org/apache/cassandra/service/MigrationCoordinator.java
@@ -341,6 +341,15 @@ public class MigrationCoordinator
         }
     }
 
+    public synchronized void removeVersionInfoForEndpoint(InetAddress endpoint)
+    {
+        Set<UUID> versions = ImmutableSet.copyOf(versionInfo.keySet());
+        for (UUID version : versions)
+        {
+            removeEndpointFromVersion(endpoint, version);
+        }
+    } 
+
     Future<Void> scheduleSchemaPull(InetAddress endpoint, VersionInfo info)
     {
         FutureTask<Void> task = new FutureTask<>(() -> pullSchema(new Callback(endpoint, info)), null);
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 8a92d72d05..5d1f83e325 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -898,12 +898,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
             return;
 
         logger.warn(String.format("There are nodes in the cluster with a different schema version than us we did not merged schemas from, " +
-                                  "our version : (%s), outstanding versions -> endpoints : %s",
+                                  "our version : (%s), outstanding versions -> endpoints : %s. Use -Dcassandra.skip_schema_check=true " +
+                                  "to ignore this.",
                                   Schema.instance.getVersion(),
                                   MigrationCoordinator.instance.outstandingVersions()));
 
         if (REQUIRE_SCHEMAS)
-            throw new RuntimeException("Didn't receive schemas for all known versions within the timeout");
+            throw new RuntimeException("Didn't receive schemas for all known versions within the timeout. " +
+                                       "Use -Dcassandra.skip_schema_check=true to skip this check.");
     }
 
     @VisibleForTesting
@@ -2461,6 +2463,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     private void removeEndpoint(InetAddress endpoint)
     {
         Gossiper.runInGossipStageBlocking(() -> Gossiper.instance.removeEndpoint(endpoint));
+        MigrationCoordinator.instance.removeVersionInfoForEndpoint(endpoint);
         SystemKeyspace.removeEndpoint(endpoint);
     }
 
@@ -2652,7 +2655,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         {
             onChange(endpoint, entry.getKey(), entry.getValue());
         }
-        MigrationCoordinator.instance.reportEndpointVersion(endpoint, epState);
     }
 
     public void onAlive(InetAddress endpoint, EndpointState state)
diff --git a/test/unit/org/apache/cassandra/service/MigrationCoordinatorTest.java b/test/unit/org/apache/cassandra/service/MigrationCoordinatorTest.java
index 6817e4ff4f..3450c2c0ae 100644
--- a/test/unit/org/apache/cassandra/service/MigrationCoordinatorTest.java
+++ b/test/unit/org/apache/cassandra/service/MigrationCoordinatorTest.java
@@ -191,6 +191,27 @@ public class MigrationCoordinatorTest
         Assert.assertTrue(signal.isSignalled());
     }
 
+	/**
+	 * If an endpoint is removed and no other endpoints are reporting its
+	 * schema version, the version should be removed and we should signal
+	 * anyone waiting on that version
+	 */
+	@Test
+	public void versionsAreSignaledWhenEndpointsRemoved()
+	{
+		InstrumentedCoordinator coordinator = new InstrumentedCoordinator();
+
+		coordinator.reportEndpointVersion(EP1, V1);
+		WaitQueue.Signal signal = coordinator.getVersionInfoUnsafe(V1).register();
+		Assert.assertFalse(signal.isSignalled());
+
+		coordinator.removeVersionInfoForEndpoint(EP1);
+		Assert.assertNull(coordinator.getVersionInfoUnsafe(V1));
+
+		Assert.assertTrue(signal.isSignalled());
+	}
+
+
     private static void assertNoContact(InstrumentedCoordinator coordinator, InetAddress endpoint, UUID version, boolean startupShouldBeUnblocked)
     {
         Assert.assertTrue(coordinator.requests.isEmpty());