Make hosts with same Endpoint share same ConvictionPolicy instance

Due to `Connection` relying on `Endpoint` to get `ConvictionPolicy` instance,
there is possibility that same connection can address different
`ConvictionPolicy` instances thoughout it's lifecycle.
As result, a connection can increase `openConnections` on one
`ConvictionPolicy` instance and decrease on another.
When it happens, assertion panic occures:
```
java.lang.AssertionError
at com.datastax.driver.core.ConvictionPolicy$DefaultConvictionPolicy.signalConnectionClosed(ConvictionPolicy.java:90)
at com.datastax.driver.core.Connection.closeAsync(Connection.java:953)
at com.datastax.driver.core.HostConnectionPool.discardAvailableConnections(HostConnectionPool.java:882)
at com.datastax.driver.core.HostConnectionPool.closeAsync(HostConnectionPool.java:843)
```

To fix that we need to make `ConvictionPolicy` instance being shared
between different hosts with same `Endpoint`.
It is not the best solution, but given lot's of code uses `Endpoint` as
unique host identifier it is good enough.

Author: dkropachev

driver-core/src/main/java/com/datastax/driver/core/Cluster.java

@@ -1611,8 +1611,7 @@ class Manager implements Connection.DefaultResponseHandler {
     Connection.Factory connectionFactory;
     ControlConnection controlConnection;
 
-    final ConvictionPolicy.Factory convictionPolicyFactory =
-        new ConvictionPolicy.DefaultConvictionPolicy.Factory();
+    final ConvictionPolicy.Factory convictionPolicyFactory;
 
     ListeningExecutorService executor;
     ListeningExecutorService blockingExecutor;
@@ -1676,8 +1675,11 @@ private Manager(
                 .withNettyOptions(configuration.getNettyOptions())
                 .withCodecRegistry(configuration.getCodecRegistry())
                 .build();
+        this.convictionPolicyFactory =
+            new ConvictionPolicy.DefaultConvictionPolicy.Factory(policies.getReconnectionPolicy());
       } else {
         this.configuration = configuration;
+        this.convictionPolicyFactory = new ConvictionPolicy.DefaultConvictionPolicy.Factory(null);
       }
       this.contactPoints = contactPoints;
       this.listeners = new CopyOnWriteArraySet<Host.StateListener>(listeners);

driver-core/src/main/java/com/datastax/driver/core/Connection.java

@@ -809,7 +809,9 @@ else if (Host.statesLogger.isDebugEnabled())
         // we still want to signal the error to the conviction policy.
         boolean decrement = signaled.compareAndSet(false, true);
 
-        boolean hostDown = host.convictionPolicy.signalConnectionFailure(this, decrement);
+        boolean hostDown =
+            host.convictionPolicy.signalConnectionFailure(
+                this, decrement, host.state == Host.State.DOWN);
         if (hostDown) {
           factory.manager.signalHostDown(host, host.wasJustAdded());
         } else {

driver-core/src/main/java/com/datastax/driver/core/ConvictionPolicy.java

@@ -19,6 +19,8 @@
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 
 import com.datastax.driver.core.policies.ReconnectionPolicy;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -45,7 +47,8 @@ abstract class ConvictionPolicy {
    *
    * @return whether the host should be considered down.
    */
-  abstract boolean signalConnectionFailure(Connection connection, boolean decrement);
+  abstract boolean signalConnectionFailure(
+      Connection connection, boolean decrement, boolean hostIsDown);
 
   abstract boolean canReconnectNow();
 
@@ -57,92 +60,111 @@ interface Factory {
     /**
      * Creates a new ConvictionPolicy instance for {@code host}.
      *
-     * @param host the host this policy applies to
+     * @param endpoint the host this policy applies to
      * @return the newly created {@link ConvictionPolicy} instance.
      */
-    ConvictionPolicy create(Host host, ReconnectionPolicy reconnectionPolicy);
+    ConvictionPolicy create(EndPoint endpoint);
   }
 
   static class DefaultConvictionPolicy extends ConvictionPolicy {
-    private final Host host;
+    private final EndPoint endpoint;
     private final ReconnectionPolicy reconnectionPolicy;
     private final AtomicInteger openConnections = new AtomicInteger();
 
     private volatile long nextReconnectionTime = Long.MIN_VALUE;
     private ReconnectionPolicy.ReconnectionSchedule reconnectionSchedule;
 
-    private DefaultConvictionPolicy(Host host, ReconnectionPolicy reconnectionPolicy) {
-      this.host = host;
+    private DefaultConvictionPolicy(EndPoint endpoint, ReconnectionPolicy reconnectionPolicy) {
+      this.endpoint = endpoint;
       this.reconnectionPolicy = reconnectionPolicy;
     }
 
     @Override
     void signalConnectionsOpening(int count) {
-      int newTotal = openConnections.addAndGet(count);
+      int newTotal = this.openConnections.addAndGet(count);
       Host.statesLogger.debug(
-          "[{}] preparing to open {} new connections, total = {}", host, count, newTotal);
+          "[{}] preparing to open {} new connections, total = {}", endpoint, count, newTotal);
       resetReconnectionTime();
     }
 
     @Override
     void signalConnectionClosed(Connection connection) {
-      int remaining = openConnections.decrementAndGet();
+      int remaining = this.openConnections.decrementAndGet();
       assert remaining >= 0;
-      Host.statesLogger.debug("[{}] {} closed, remaining = {}", host, connection, remaining);
+      Host.statesLogger.debug("[{}] {} closed, remaining = {}", endpoint, connection, remaining);
     }
 
     @Override
-    boolean signalConnectionFailure(Connection connection, boolean decrement) {
+    boolean signalConnectionFailure(Connection connection, boolean decrement, boolean hostIsDown) {
       int remaining;
       if (decrement) {
-        if (host.state != Host.State.DOWN) updateReconnectionTime();
+        if (!hostIsDown) updateReconnectionTime();
 
-        remaining = openConnections.decrementAndGet();
+        remaining = this.openConnections.decrementAndGet();
         assert remaining >= 0;
-        Host.statesLogger.debug("[{}] {} failed, remaining = {}", host, connection, remaining);
+        Host.statesLogger.debug("[{}] {} failed, remaining = {}", endpoint, connection, remaining);
       } else {
-        remaining = openConnections.get();
+        remaining = this.openConnections.get();
       }
       return remaining == 0;
     }
 
     private synchronized void updateReconnectionTime() {
       long now = System.nanoTime();
-      if (nextReconnectionTime > now)
+      if (this.nextReconnectionTime > now)
         // Someone else updated the time before us
         return;
 
-      if (reconnectionSchedule == null) reconnectionSchedule = reconnectionPolicy.newSchedule();
+      if (this.reconnectionSchedule == null)
+        this.reconnectionSchedule = reconnectionPolicy.newSchedule();
 
-      long nextDelayMs = reconnectionSchedule.nextDelayMs();
+      long nextDelayMs = this.reconnectionSchedule.nextDelayMs();
       Host.statesLogger.debug(
-          "[{}] preventing new connections for the next {} ms", host, nextDelayMs);
-      nextReconnectionTime = now + NANOSECONDS.convert(nextDelayMs, MILLISECONDS);
+          "[{}] preventing new connections for the next {} ms", endpoint, nextDelayMs);
+      this.nextReconnectionTime = now + NANOSECONDS.convert(nextDelayMs, MILLISECONDS);
     }
 
     private synchronized void resetReconnectionTime() {
-      reconnectionSchedule = null;
-      nextReconnectionTime = Long.MIN_VALUE;
+      this.reconnectionSchedule = null;
+      this.nextReconnectionTime = Long.MIN_VALUE;
     }
 
     @Override
     boolean canReconnectNow() {
       boolean canReconnectNow =
-          nextReconnectionTime == Long.MIN_VALUE || System.nanoTime() >= nextReconnectionTime;
+          this.nextReconnectionTime == Long.MIN_VALUE
+              || System.nanoTime() >= this.nextReconnectionTime;
       Host.statesLogger.trace("canReconnectNow={}", canReconnectNow);
       return canReconnectNow;
     }
 
+    private static class ConvictionState {}
+
     @Override
     boolean hasActiveConnections() {
-      return openConnections.get() > 0;
+      return this.openConnections.get() > 0;
     }
 
     static class Factory implements ConvictionPolicy.Factory {
+      private final ConcurrentMap<EndPoint, ConvictionPolicy> policyCache =
+          new ConcurrentHashMap<>();
+      private final ReconnectionPolicy reconnectionPolicy;
+
+      public Factory(ReconnectionPolicy reconnectionPolicy) {
+        this.reconnectionPolicy = reconnectionPolicy;
+      }
 
       @Override
-      public ConvictionPolicy create(Host host, ReconnectionPolicy reconnectionPolicy) {
-        return new DefaultConvictionPolicy(host, reconnectionPolicy);
+      public ConvictionPolicy create(EndPoint endpoint) {
+        ConvictionPolicy convictionPolicy = policyCache.get(endpoint);
+        if (convictionPolicy == null) {
+          convictionPolicy = new DefaultConvictionPolicy(endpoint, reconnectionPolicy);
+          ConvictionPolicy alreadyExisting = policyCache.putIfAbsent(endpoint, convictionPolicy);
+          if (alreadyExisting != null) {
+            return alreadyExisting;
+          }
+        }
+        return convictionPolicy;
       }
     }
   }

driver-core/src/main/java/com/datastax/driver/core/DelegatingCluster.java

@@ -44,8 +44,6 @@ protected DelegatingCluster() {
     // accidentally calling a
     // parent method could initialize the parent state.
 
-    // Construct parent class with dummy parameters that will never get used (since super.init() is
-    // never called).
     super("delegating_cluster", Collections.<EndPoint>emptyList(), null);
 
     // Immediately close the parent class's internal Manager, to make sure that it will fail fast if

driver-core/src/main/java/com/datastax/driver/core/Host.java

@@ -109,7 +109,7 @@ enum State {
     if (endPoint == null || convictionPolicyFactory == null) throw new NullPointerException();
 
     this.endPoint = endPoint;
-    this.convictionPolicy = convictionPolicyFactory.create(this, manager.reconnectionPolicy());
+    this.convictionPolicy = convictionPolicyFactory.create(this.endPoint);
     this.manager = manager;
     this.defaultExecutionInfo = new ExecutionInfo(this);
     this.state = State.ADDED;

driver-core/src/test/java/com/datastax/driver/core/AbstractReplicationStrategyTest.java

@@ -45,7 +45,7 @@ protected static class HostMock extends Host {
     private HostMock(InetSocketAddress address, Cluster.Manager manager) {
       super(
           EndPoints.forAddress(address),
-          new ConvictionPolicy.DefaultConvictionPolicy.Factory(),
+          new ConvictionPolicy.DefaultConvictionPolicy.Factory(null),
           manager);
       this.address = address;
     }