JAVA-1877: Use a separate reconnection schedule for the control connection

Author: olim7t

changelog/README.md

@@ -4,6 +4,7 @@
 
 ### 4.0.0-beta1 (in progress)
 
+- [improvement] JAVA-1877: Use a separate reconnection schedule for the control connection
 - [improvement] JAVA-1763: Generate a binary tarball as part of the build process
 - [improvement] JAVA-1884: Add additional methods from TypeToken to GenericType
 - [improvement] JAVA-1883: Use custom queue implementation for LBP's query plan

core/src/main/java/com/datastax/oss/driver/api/core/connection/ReconnectionPolicy.java

@@ -15,23 +15,41 @@
  */
 package com.datastax.oss.driver.api.core.connection;
 
+import com.datastax.oss.driver.api.core.metadata.Node;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.time.Duration;
 
 /**
- * Decides how often the driver tries to re-establish lost connections to a node.
+ * Decides how often the driver tries to re-establish lost connections.
  *
- * <p>Each time a connection to a node is lost, a {@link #newSchedule() new schedule} instance gets
- * created. Then {@link ReconnectionSchedule#nextDelay()} will be called each time the driver needs
- * to schedule the next connection attempt. When the node is back to its required number of
- * connections, the schedule will be reset (that is, the next failure will create a fresh schedule
- * instance).
+ * <p>Each time a reconnection starts, a new {@link ReconnectionSchedule} instance gets created.
+ * Then {@link ReconnectionSchedule#nextDelay()} will be called each time the driver needs to
+ * program the next connection attempt. When the reconnection succeeds, the schedule is discarded,
+ * and the next reconnection will start from a fresh instance.
+ *
+ * <p>There are two types of reconnection:
+ *
+ * <ul>
+ *   <li>{@linkplain #newNodeSchedule(Node) for regular node connections}: when the connection pool
+ *       for a node does not have its configured number of connections (see {@code
+ *       advanced.connection.pool.*.size} in the configuration), a reconnection starts for that
+ *       pool.
+ *   <li>{@linkplain #newControlConnectionSchedule() for the control connection}: when the control
+ *       node goes down, a reconnection starts to find another node to replace it.
+ * </ul>
+ *
+ * This interface defines separate methods for those two cases, but implementations are free to
+ * delegate to the same method internally if the same type of schedule can be used.
  */
 public interface ReconnectionPolicy extends AutoCloseable {
 
-  /** Creates a new {@linkplain ReconnectionSchedule schedule}. */
+  /** Creates a new schedule for the given node. */
+  @NonNull
+  ReconnectionSchedule newNodeSchedule(@NonNull Node node);
+
+  /** Creates a new schedule for the control connection. */
   @NonNull
-  ReconnectionSchedule newSchedule();
+  ReconnectionSchedule newControlConnectionSchedule();
 
   /** Called when the cluster that this policy is associated with closes. */
   @Override

core/src/main/java/com/datastax/oss/driver/internal/core/connection/ConstantReconnectionPolicy.java

@@ -19,16 +19,23 @@
 import com.datastax.oss.driver.api.core.config.DriverConfigProfile;
 import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy;
 import com.datastax.oss.driver.api.core.context.DriverContext;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** A reconnection policy that waits a constant time between each reconnection attempt. */
 public class ConstantReconnectionPolicy implements ReconnectionPolicy {
 
+  private static final Logger LOG = LoggerFactory.getLogger(ConstantReconnectionPolicy.class);
+
+  private final String logPrefix;
   private final ReconnectionSchedule schedule;
 
   /** Builds a new instance. */
   public ConstantReconnectionPolicy(DriverContext context) {
+    this.logPrefix = context.sessionName();
     DriverConfigProfile config = context.config().getDefaultProfile();
     Duration delay = config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY);
     if (delay.isNegative()) {
@@ -44,7 +51,15 @@ public ConstantReconnectionPolicy(DriverContext context) {
 
   @NonNull
   @Override
-  public ReconnectionSchedule newSchedule() {
+  public ReconnectionSchedule newNodeSchedule(@NonNull Node node) {
+    LOG.debug("[{}] Creating new schedule for {}", logPrefix, node);
+    return schedule;
+  }
+
+  @NonNull
+  @Override
+  public ReconnectionSchedule newControlConnectionSchedule() {
+    LOG.debug("[{}] Creating new schedule for the control connection", logPrefix);
     return schedule;
   }
 

core/src/main/java/com/datastax/oss/driver/internal/core/connection/ExponentialReconnectionPolicy.java

@@ -19,24 +19,36 @@
 import com.datastax.oss.driver.api.core.config.DriverConfigProfile;
 import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy;
 import com.datastax.oss.driver.api.core.context.DriverContext;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.shaded.guava.common.base.Preconditions;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.time.Duration;
 import net.jcip.annotations.ThreadSafe;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A reconnection policy that waits exponentially longer between each reconnection attempt (but
  * keeps a constant delay once a maximum delay is reached).
+ *
+ * <p>It uses the same schedule implementation for individual nodes or the control connection:
+ * reconnection attempt {@code i} will be tried {@code Math.min(2^(i-1) * getBaseDelayMs(),
+ * getMaxDelayMs())} milliseconds after the previous one.
  */
 @ThreadSafe
 public class ExponentialReconnectionPolicy implements ReconnectionPolicy {
 
+  private static final Logger LOG = LoggerFactory.getLogger(ExponentialReconnectionPolicy.class);
+
+  private final String logPrefix;
   private final long baseDelayMs;
   private final long maxDelayMs;
   private final long maxAttempts;
 
   /** Builds a new instance. */
   public ExponentialReconnectionPolicy(DriverContext context) {
+    this.logPrefix = context.sessionName();
+
     DriverConfigProfile config = context.config().getDefaultProfile();
 
     this.baseDelayMs = config.getDuration(DefaultDriverOption.RECONNECTION_BASE_DELAY).toMillis();
@@ -84,17 +96,17 @@ public long getMaxDelayMs() {
     return maxDelayMs;
   }
 
-  /**
-   * A new schedule that used an exponentially growing delay between reconnection attempts.
-   *
-   * <p>For this schedule, reconnection attempt {@code i} will be tried {@code Math.min(2^(i-1) *
-   * getBaseDelayMs(), getMaxDelayMs())} milliseconds after the previous one.
-   *
-   * @return the newly created schedule.
-   */
   @NonNull
   @Override
-  public ReconnectionSchedule newSchedule() {
+  public ReconnectionSchedule newNodeSchedule(@NonNull Node node) {
+    LOG.debug("[{}] Creating new schedule for {}", logPrefix, node);
+    return new ExponentialSchedule();
+  }
+
+  @NonNull
+  @Override
+  public ReconnectionSchedule newControlConnectionSchedule() {
+    LOG.debug("[{}] Creating new schedule for the control connection", logPrefix);
     return new ExponentialSchedule();
   }
 

core/src/main/java/com/datastax/oss/driver/internal/core/control/ControlConnection.java

@@ -17,6 +17,7 @@
 
 import com.datastax.oss.driver.api.core.AllNodesFailedException;
 import com.datastax.oss.driver.api.core.AsyncAutoCloseable;
+import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy;
 import com.datastax.oss.driver.api.core.loadbalancing.NodeDistance;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.api.core.metadata.NodeState;
@@ -224,8 +225,13 @@ private class SingleThreaded {
 
     private SingleThreaded(InternalDriverContext context) {
       this.context = context;
+      ReconnectionPolicy reconnectionPolicy = context.reconnectionPolicy();
       this.reconnection =
-          new Reconnection(logPrefix, adminExecutor, context.reconnectionPolicy(), this::reconnect);
+          new Reconnection(
+              logPrefix,
+              adminExecutor,
+              reconnectionPolicy::newControlConnectionSchedule,
+              this::reconnect);
       // In "reconnect-on-init" mode, handle cancellation of the initFuture by user code
       CompletableFutures.whenCancelled(
           this.initFuture,

core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java

@@ -22,6 +22,7 @@
 import com.datastax.oss.driver.api.core.auth.AuthenticationException;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverConfig;
+import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy;
 import com.datastax.oss.driver.api.core.loadbalancing.NodeDistance;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.api.core.metrics.DefaultNodeMetric;
@@ -234,11 +235,12 @@ private SingleThreaded(
       this.wantedCount = getConfiguredSize(distance);
       this.channelFactory = context.channelFactory();
       this.eventBus = context.eventBus();
+      ReconnectionPolicy reconnectionPolicy = context.reconnectionPolicy();
       this.reconnection =
           new Reconnection(
               logPrefix,
               adminExecutor,
-              context.reconnectionPolicy(),
+              () -> reconnectionPolicy.newNodeSchedule(node),
               this::addMissingChannels,
               () -> eventBus.fire(ChannelEvent.reconnectionStarted(node)),
               () -> eventBus.fire(ChannelEvent.reconnectionStopped(node)));

core/src/main/java/com/datastax/oss/driver/internal/core/util/concurrent/Reconnection.java

@@ -15,7 +15,7 @@
  */
 package com.datastax.oss.driver.internal.core.util.concurrent;
 
-import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy;
+import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy.ReconnectionSchedule;
 import com.datastax.oss.driver.internal.core.util.Loggers;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.Future;
@@ -25,6 +25,7 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 import net.jcip.annotations.NotThreadSafe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -50,13 +51,13 @@ private enum State {
 
   private final String logPrefix;
   private final EventExecutor executor;
-  private final ReconnectionPolicy reconnectionPolicy;
+  private final Supplier<ReconnectionSchedule> scheduleSupplier;
   private final Callable<CompletionStage<Boolean>> reconnectionTask;
   private final Runnable onStart;
   private final Runnable onStop;
 
   private State state = State.STOPPED;
-  private ReconnectionPolicy.ReconnectionSchedule reconnectionSchedule;
+  private ReconnectionSchedule reconnectionSchedule;
   private ScheduledFuture<CompletionStage<Boolean>> nextAttempt;
 
   /**
@@ -66,13 +67,13 @@ private enum State {
   public Reconnection(
       String logPrefix,
       EventExecutor executor,
-      ReconnectionPolicy reconnectionPolicy,
+      Supplier<ReconnectionSchedule> scheduleSupplier,
       Callable<CompletionStage<Boolean>> reconnectionTask,
       Runnable onStart,
       Runnable onStop) {
     this.logPrefix = logPrefix;
     this.executor = executor;
-    this.reconnectionPolicy = reconnectionPolicy;
+    this.scheduleSupplier = scheduleSupplier;
     this.reconnectionTask = reconnectionTask;
     this.onStart = onStart;
     this.onStop = onStop;
@@ -81,9 +82,9 @@ public Reconnection(
   public Reconnection(
       String logPrefix,
       EventExecutor executor,
-      ReconnectionPolicy reconnectionPolicy,
+      Supplier<ReconnectionSchedule> scheduleSupplier,
       Callable<CompletionStage<Boolean>> reconnectionTask) {
-    this(logPrefix, executor, reconnectionPolicy, reconnectionTask, () -> {}, () -> {});
+    this(logPrefix, executor, scheduleSupplier, reconnectionTask, () -> {}, () -> {});
   }
 
   /**
@@ -108,7 +109,7 @@ public void start() {
         state = State.ATTEMPT_IN_PROGRESS;
         break;
       case STOPPED:
-        reconnectionSchedule = reconnectionPolicy.newSchedule();
+        reconnectionSchedule = scheduleSupplier.get();
         onStart.run();
         scheduleNextAttempt();
         break;
@@ -180,7 +181,7 @@ private void scheduleNextAttempt() {
     assert executor.inEventLoop();
     state = State.SCHEDULED;
     if (reconnectionSchedule == null) { // happens if reconnectNow() while we were stopped
-      reconnectionSchedule = reconnectionPolicy.newSchedule();
+      reconnectionSchedule = scheduleSupplier.get();
     }
     Duration nextInterval = reconnectionSchedule.nextDelay();
     LOG.debug("[{}] Scheduling next reconnection in {}", logPrefix, nextInterval);

core/src/test/java/com/datastax/oss/driver/internal/core/control/ControlConnectionTest.java

@@ -16,9 +16,11 @@
 package com.datastax.oss.driver.internal.core.control;
 
 import static com.datastax.oss.driver.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.never;
 
 import com.datastax.oss.driver.api.core.loadbalancing.NodeDistance;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.api.core.metadata.NodeState;
 import com.datastax.oss.driver.internal.core.channel.ChannelEvent;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
@@ -105,7 +107,7 @@ public void should_init_with_second_contact_point_if_first_one_fails() {
     Mockito.verify(eventBus).fire(ChannelEvent.controlConnectionFailed(node1));
     Mockito.verify(eventBus).fire(ChannelEvent.channelOpened(node2));
     // each attempt tries all nodes, so there is no reconnection
-    Mockito.verify(reconnectionPolicy, never()).newSchedule();
+    Mockito.verify(reconnectionPolicy, never()).newNodeSchedule(any(Node.class));
 
     factoryHelper.verifyNoMoreCalls();
   }
@@ -130,7 +132,7 @@ public void should_fail_to_init_if_all_contact_points_fail() {
     Mockito.verify(eventBus).fire(ChannelEvent.controlConnectionFailed(node1));
     Mockito.verify(eventBus).fire(ChannelEvent.controlConnectionFailed(node2));
     // no reconnections at init
-    Mockito.verify(reconnectionPolicy, never()).newSchedule();
+    Mockito.verify(reconnectionPolicy, never()).newNodeSchedule(any(Node.class));
 
     factoryHelper.verifyNoMoreCalls();
   }

core/src/test/java/com/datastax/oss/driver/internal/core/control/ControlConnectionTestBase.java

@@ -95,7 +95,8 @@ public void setup() {
             });
 
     Mockito.when(context.reconnectionPolicy()).thenReturn(reconnectionPolicy);
-    Mockito.when(reconnectionPolicy.newSchedule()).thenReturn(reconnectionSchedule);
+    Mockito.when(reconnectionPolicy.newControlConnectionSchedule())
+        .thenReturn(reconnectionSchedule);
     // By default, set a large reconnection delay. Tests that care about reconnection will override
     // it.
     Mockito.when(reconnectionSchedule.nextDelay()).thenReturn(Duration.ofDays(1));

core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolTestBase.java

@@ -22,6 +22,7 @@
 import com.datastax.oss.driver.api.core.config.DriverConfig;
 import com.datastax.oss.driver.api.core.config.DriverConfigProfile;
 import com.datastax.oss.driver.api.core.connection.ReconnectionPolicy;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.channel.ChannelFactory;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
 import com.datastax.oss.driver.internal.core.context.EventBus;
@@ -76,7 +77,8 @@ public void setup() {
     Mockito.when(context.channelFactory()).thenReturn(channelFactory);
 
     Mockito.when(context.reconnectionPolicy()).thenReturn(reconnectionPolicy);
-    Mockito.when(reconnectionPolicy.newSchedule()).thenReturn(reconnectionSchedule);
+    Mockito.when(reconnectionPolicy.newNodeSchedule(any(Node.class)))
+        .thenReturn(reconnectionSchedule);
     // By default, set a large reconnection delay. Tests that care about reconnection will override
     // it.
     Mockito.when(reconnectionSchedule.nextDelay()).thenReturn(Duration.ofDays(1));