Make sure topology events are sent to clients when using a single network interface

Patch by Alan Boudrealt and Bryn Cooke; reviewed by brandonwilliams for CASSANDRA-15677

CHANGES.txt

 4.0-alpha5
+ * Fix missing topology events when running multiple nodes on the same network interface (CASSANDRA-15677)
  * Create config.yml.MIDRES (CASSANDRA-15712)
  * Fix handling of fully purged static rows in repaired data tracking (CASSANDRA-15848)
  * Prevent validation request submission from blocking ANTI_ENTROPY stage (CASSANDRA-15812)

build.xml

@@ -633,6 +633,7 @@
           <dependency groupId="com.beust" artifactId="jcommander" version="1.30"/>
           <!-- when updating assertj, make sure to also update the corresponding junit-bom dependency -->
           <dependency groupId="org.assertj" artifactId="assertj-core" version="3.15.0"/>
+          <dependency groupId="org.awaitility" artifactId="awaitility" version="4.0.3" />
 
         </dependencyManagement>
         <developer id="adelapena" name="Andres de la Peña"/>
@@ -718,6 +719,7 @@
              this that the new assertj's `assertj-parent-pom` depends on. -->
         <dependency groupId="org.junit" artifactId="junit-bom" version="5.6.0" type="pom"/>
         <dependency groupId="org.assertj" artifactId="assertj-core"/>
+        <dependency groupId="org.awaitility" artifactId="awaitility"/>
       </artifact:pom>
       <!-- this build-deps-pom-sources "artifact" is the same as build-deps-pom but only with those
            artifacts that have "-source.jar" files -->
@@ -737,6 +739,7 @@
         <dependency groupId="net.ju-n.compile-command-annotations" artifactId="compile-command-annotations"/>
         <dependency groupId="org.apache.ant" artifactId="ant-junit" version="1.9.7" />
         <dependency groupId="org.assertj" artifactId="assertj-core"/>
+        <dependency groupId="org.awaitility" artifactId="awaitility"/>
       </artifact:pom>
 
       <artifact:pom id="coverage-deps-pom"

src/java/org/apache/cassandra/transport/Event.java

@@ -90,9 +90,9 @@ public abstract class Event
     {
         public final InetSocketAddress node;
 
-        public InetAddress nodeAddress()
+        public InetAddressAndPort nodeAddressAndPort()
         {
-            return node.getAddress();
+            return InetAddressAndPort.getByAddressOverrideDefaults(node.getAddress(), node.getPort());
         }
 
         private NodeEvent(Type type, InetSocketAddress node)

src/java/org/apache/cassandra/transport/Server.java

@@ -613,7 +613,7 @@ public class Server implements CassandraDaemon.Server
         private void send(InetAddressAndPort endpoint, Event.NodeEvent event)
         {
             if (logger.isTraceEnabled())
-                logger.trace("Sending event for endpoint {}, rpc address {}", endpoint, event.nodeAddress());
+                logger.trace("Sending event for endpoint {}, rpc address {}", endpoint, event.nodeAddressAndPort());
 
             // If the endpoint is not the local node, extract the node address
             // and if it is the same as our own RPC broadcast address (which defaults to the rcp address)
@@ -621,7 +621,7 @@ public class Server implements CassandraDaemon.Server
             // which is not useful to any driver and in fact may cauase serious problems to some drivers,
             // see CASSANDRA-10052
             if (!endpoint.equals(FBUtilities.getBroadcastAddressAndPort()) &&
-                event.nodeAddress().equals(FBUtilities.getJustBroadcastNativeAddress()))
+                event.nodeAddressAndPort().equals(FBUtilities.getBroadcastNativeAddressAndPort()))
                 return;
 
             send(event);

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.datastax.driver.core.Session;
+import org.apache.cassandra.distributed.Cluster;
+
+import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
+import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL;
+import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.apache.cassandra.distributed.impl.INodeProvisionStrategy.Strategy.OneNetworkInterface;
+import static org.awaitility.Awaitility.await;
+
+public class NodeDecommissionTest extends TestBaseImpl
+{
+
+    @Test
+    public void testDecomissionSucceedsForNodesOnTheSameInterface() throws Throwable
+    {
+        try (Cluster control = init(Cluster.build().withNodes(3).withNodeProvisionStrategy(OneNetworkInterface).withConfig(
+        config -> {
+            config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL);
+        }).start()))
+        {
+            final com.datastax.driver.core.Cluster cluster = com.datastax.driver.core.Cluster.builder().addContactPoint("127.0.0.1").build();
+            Session session = cluster.connect();
+            control.get(3).nodetool("disablebinary");
+            control.get(3).nodetool("decommission", "-f");
+            await().atMost(10, TimeUnit.SECONDS)
+                   .untilAsserted(() -> Assert.assertEquals(2, cluster.getMetadata().getAllHosts().size()));
+            session.close();
+            cluster.close();
+        }
+    }
+}
+