Update driver version to prevent issues with extra events being received when a node is decommissioned.

Patch by Bryn Cooke, reviewed by brandonwilliams for CASSANDRA-15677

build.xml

@@ -599,7 +599,7 @@
 	  </dependency>
           <dependency groupId="com.google.code.findbugs" artifactId="jsr305" version="2.0.2" />
           <dependency groupId="com.clearspring.analytics" artifactId="stream" version="2.5.2" />
-          <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" version="3.6.0" classifier="shaded">
+          <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" version="3.9.0" classifier="shaded">
             <exclusion groupId="io.netty" artifactId="netty-buffer"/>
             <exclusion groupId="io.netty" artifactId="netty-codec"/>
             <exclusion groupId="io.netty" artifactId="netty-handler"/>

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

@@ -1,57 +0,0 @@
-/*
- * 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();
-        }
-    }
-}
-

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

@@ -0,0 +1,198 @@
+/*
+ * 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.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import com.datastax.driver.core.Host;
+import com.datastax.driver.core.Session;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.impl.INodeProvisionStrategy.Strategy;
+import org.apache.cassandra.distributed.test.TopologyChangeTest.EventStateListener.Event;
+
+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.MultipleNetworkInterfaces;
+import static org.apache.cassandra.distributed.impl.INodeProvisionStrategy.Strategy.OneNetworkInterface;
+import static org.apache.cassandra.distributed.test.TopologyChangeTest.EventStateListener.EventType.Down;
+import static org.apache.cassandra.distributed.test.TopologyChangeTest.EventStateListener.EventType.Remove;
+import static org.apache.cassandra.distributed.test.TopologyChangeTest.EventStateListener.EventType.Up;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+@RunWith(Parameterized.class)
+public class TopologyChangeTest extends TestBaseImpl
+{
+    static class EventStateListener implements Host.StateListener
+    {
+        enum EventType
+        {
+            Add,
+            Up,
+            Down,
+            Remove,
+        }
+
+        static class Event
+        {
+            String host;
+            EventType type;
+
+            Event(EventType type, Host host)
+            {
+                this.type = type;
+                this.host = host.getBroadcastSocketAddress().toString();
+            }
+
+            public Event(EventType type, IInvokableInstance iInvokableInstance)
+            {
+                this.type = type;
+                this.host = iInvokableInstance.broadcastAddress().toString();
+            }
+
+
+            public String toString()
+            {
+                return "Event{" +
+                       "host='" + host + '\'' +
+                       ", type=" + type +
+                       '}';
+            }
+
+            public boolean equals(Object o)
+            {
+                if (this == o) return true;
+                if (o == null || getClass() != o.getClass()) return false;
+                Event event = (Event) o;
+                return Objects.equals(host, event.host) &&
+                       type == event.type;
+            }
+
+            public int hashCode()
+            {
+                return Objects.hash(host, type);
+            }
+        }
+
+        private List<Event> events = new ArrayList<>();
+
+        public void onAdd(Host host)
+        {
+            events.add(new Event(EventType.Add, host));
+        }
+
+        public void onUp(Host host)
+        {
+            events.add(new Event(Up, host));
+        }
+
+        public void onDown(Host host)
+        {
+            events.add(new Event(EventType.Down, host));
+        }
+
+        public void onRemove(Host host)
+        {
+            events.add(new Event(Remove, host));
+        }
+
+        public void onRegister(com.datastax.driver.core.Cluster cluster)
+        {
+        }
+
+        public void onUnregister(com.datastax.driver.core.Cluster cluster)
+        {
+        }
+    }
+
+    @Parameterized.Parameter(0)
+    public Strategy strategy;
+
+    @Parameterized.Parameters(name = "{index}: provision strategy={0}")
+    public static Collection<Strategy[]> data()
+    {
+        return Arrays.asList(new Strategy[][]{ { MultipleNetworkInterfaces },
+                                               { OneNetworkInterface }
+        });
+    }
+
+    @Test
+    public void testDecommission() throws Throwable
+    {
+        try (Cluster control = init(Cluster.build().withNodes(3).withNodeProvisionStrategy(strategy).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();
+            EventStateListener eventStateListener = new EventStateListener();
+            session.getCluster().register(eventStateListener);
+            control.get(3).nodetool("disablebinary");
+            control.get(3).nodetool("decommission", "-f");
+            await().atMost(5, TimeUnit.SECONDS)
+                   .untilAsserted(() -> Assert.assertEquals(2, cluster.getMetadata().getAllHosts().size()));
+            assertThat(eventStateListener.events).containsExactly(new Event(Remove, control.get(3)));
+            session.close();
+            cluster.close();
+        }
+    }
+
+    @Test
+    public void testRestartNode() throws Throwable
+    {
+        try (Cluster control = init(Cluster.build().withNodes(3).withNodeProvisionStrategy(strategy).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();
+            EventStateListener eventStateListener = new EventStateListener();
+            session.getCluster().register(eventStateListener);
+
+            control.get(3).shutdown().get();
+            await().atMost(5, TimeUnit.SECONDS)
+                   .untilAsserted(() -> Assert.assertEquals(2, cluster.getMetadata().getAllHosts().stream().filter(h -> h.isUp()).count()));
+
+            control.get(3).startup();
+            await().atMost(30, TimeUnit.SECONDS)
+                   .untilAsserted(() -> Assert.assertEquals(3, cluster.getMetadata().getAllHosts().stream().filter(h -> h.isUp()).count()));
+
+            assertThat(eventStateListener.events).containsExactly(new Event(Down, control.get(3)),
+                                                                  new Event(Up, control.get(3)));
+
+            session.close();
+            cluster.close();
+        }
+    }
+}
+

test/unit/org/apache/cassandra/auth/PasswordAuthenticatorTest.java

@@ -26,6 +26,7 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 import com.datastax.driver.core.Authenticator;
+import com.datastax.driver.core.EndPoint;
 import com.datastax.driver.core.PlainTextAuthProvider;
 import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.cql3.CQLTester;
@@ -121,7 +122,7 @@ public class PasswordAuthenticatorTest extends CQLTester
     {
         SaslNegotiator negotiator = authenticator.newSaslNegotiator(null);
         Authenticator clientAuthenticator = (new PlainTextAuthProvider(username, password))
-                                            .newAuthenticator(null, null);
+                                            .newAuthenticator((EndPoint) null, null);
 
         negotiator.evaluateResponse(clientAuthenticator.initialResponse());
         negotiator.getAuthenticatedUser();