Avoid NPE during cms initialization abort

Patch by marcuse; reviewed by David Capwell and Caleb Rackliffe for CASSANDRA-20527

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Avoid NPE during cms initialization abort (CASSANDRA-20527)
  * Avoid failing queries when epoch changes and replica goes up/down (CASSANDRA-20489)
  * Split out truncation record lock (CASSANDRA-20480)
  * Throw new IndexBuildInProgressException when queries fail during index build, instead of IndexNotAvailableException (CASSANDRA-20402)

src/java/org/apache/cassandra/tcm/Startup.java

@@ -246,7 +246,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
             else
             {
                 CMSInitializationRequest.Initiator initiator = Election.instance.initiator();
-                candidates = Discovery.instance.discoverOnce(initiator == null ? null : initiator.initiator);
+                candidates = Discovery.instance.discoverOnce(initiator == null ? null : initiator.endpoint);
             }
             Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
         }

src/java/org/apache/cassandra/tcm/migration/CMSInitializationRequest.java

@@ -109,12 +109,12 @@ public class CMSInitializationRequest
     public static class Initiator
     {
         public static final Serializer serializer = new Serializer();
-        public final InetAddressAndPort initiator;
+        public final InetAddressAndPort endpoint;
         public final UUID initToken;
 
         public Initiator(InetAddressAndPort initiator, UUID initToken)
         {
-            this.initiator = initiator;
+            this.endpoint = initiator;
             this.initToken = initToken;
         }
 
@@ -124,20 +124,20 @@ public class CMSInitializationRequest
             if (this == o) return true;
             if (!(o instanceof Initiator)) return false;
             Initiator other = (Initiator) o;
-            return Objects.equals(initiator, other.initiator) && Objects.equals(initToken, other.initToken);
+            return Objects.equals(endpoint, other.endpoint) && Objects.equals(initToken, other.initToken);
         }
 
         @Override
         public int hashCode()
         {
-            return Objects.hash(initiator, initToken);
+            return Objects.hash(endpoint, initToken);
         }
 
         @Override
         public String toString()
         {
             return "Initiator{" +
-                   "initiator=" + initiator +
+                   "initiator=" + endpoint +
                    ", initToken=" + initToken +
                    '}';
         }
@@ -147,7 +147,7 @@ public class CMSInitializationRequest
             @Override
             public void serialize(Initiator t, DataOutputPlus out, int version) throws IOException
             {
-                InetAddressAndPort.Serializer.inetAddressAndPortSerializer.serialize(t.initiator, out, version);
+                InetAddressAndPort.Serializer.inetAddressAndPortSerializer.serialize(t.endpoint, out, version);
                 UUIDSerializer.serializer.serialize(t.initToken, out, version);
             }
 
@@ -161,7 +161,7 @@ public class CMSInitializationRequest
             @Override
             public long serializedSize(Initiator t, int version)
             {
-                return InetAddressAndPort.Serializer.inetAddressAndPortSerializer.serializedSize(t.initiator, version) +
+                return InetAddressAndPort.Serializer.inetAddressAndPortSerializer.serializedSize(t.endpoint, version) +
                        UUIDSerializer.serializer.serializedSize(t.initToken, version);
             }
         }

src/java/org/apache/cassandra/tcm/migration/Election.java

@@ -134,7 +134,7 @@ public class Election
     {
         CMSInitializationRequest.Initiator currentInitiator = initiator.get();
         if (currentInitiator != null &&
-            Objects.equals(currentInitiator.initiator, FBUtilities.getBroadcastAddressAndPort()) &&
+            Objects.equals(currentInitiator.endpoint, FBUtilities.getBroadcastAddressAndPort()) &&
             initiator.compareAndSet(currentInitiator, MIGRATING))
         {
             Startup.initializeAsFirstCMSNode();
@@ -183,7 +183,7 @@ public class Election
     {
         InetAddressAndPort expectedInitiator = InetAddressAndPort.getByNameUnchecked(initiatorEp);
         CMSInitializationRequest.Initiator currentInitiator = initiator.get();
-        if (currentInitiator != null && Objects.equals(currentInitiator.initiator, expectedInitiator) && initiator.compareAndSet(currentInitiator, null))
+        if (currentInitiator != null && Objects.equals(currentInitiator.endpoint, expectedInitiator) && initiator.compareAndSet(currentInitiator, null))
         {
             ClusterMetadata metadata = ClusterMetadata.current();
             for (Map.Entry<NodeId, NodeState> entry : metadata.directory.states.entrySet())
@@ -243,9 +243,11 @@ public class Election
         public void doVerb(Message<CMSInitializationRequest.Initiator> message) throws IOException
         {
             logger.info("Received election abort message {} from {}", message.payload, message.from());
-            CMSInitializationRequest.Initiator initiator = message.payload;
+            CMSInitializationRequest.Initiator remoteInitiator = message.payload;
-            if (!initiator.initiator.equals(initiator().initiator) || !updateInitiator(message.payload, null))
+            if (initiator() == null)
-                logger.error("Could not clear initiator - initiator is set to {}, abort message received from {}", initiator(), message.payload);
+                logger.info("Initiator already cleared, ignoring abort message from {}: {}", message.from(), remoteInitiator);
+            else if (!remoteInitiator.endpoint.equals(initiator().endpoint) || !updateInitiator(remoteInitiator, null))
+                logger.error("Could not clear initiator - initiator is set to {}, abort message received from {}: {}", initiator(), message.from(), remoteInitiator);
         }
     }
 }