JAVA-2951: Accept multiple node state listeners, schema change listeners and request trackers (#1565)

Author: adutra

changelog/README.md

@@ -4,6 +4,8 @@
 
 ### 4.13.0 (in progress)
 
+- [improvement] JAVA-2951: Accept multiple node state listeners, schema change listeners and request trackers
+
 Merged from 4.12.x:
 
 - [bug] JAVA-2949: Provide mapper support for CompletionStage<Stream<T>>

core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java

@@ -296,7 +296,10 @@ public enum DefaultDriverOption implements DriverOption {
    * The class of a session-wide component that tracks the outcome of requests.
    *
    * <p>Value-type: {@link String}
+   *
+   * @deprecated Use {@link #REQUEST_TRACKER_CLASSES} instead.
    */
+  @Deprecated
   REQUEST_TRACKER_CLASS("advanced.request-tracker.class"),
   /**
    * Whether to log successful requests.
@@ -388,14 +391,20 @@ public enum DefaultDriverOption implements DriverOption {
    * The class of a session-wide component that listens for node state changes.
    *
    * <p>Value-type: {@link String}
+   *
+   * @deprecated Use {@link #METADATA_NODE_STATE_LISTENER_CLASSES} instead.
    */
+  @Deprecated
   METADATA_NODE_STATE_LISTENER_CLASS("advanced.node-state-listener.class"),
 
   /**
    * The class of a session-wide component that listens for schema changes.
    *
    * <p>Value-type: {@link String}
+   *
+   * @deprecated Use {@link #METADATA_SCHEMA_CHANGE_LISTENER_CLASSES} instead.
    */
+  @Deprecated
   METADATA_SCHEMA_CHANGE_LISTENER_CLASS("advanced.schema-change-listener.class"),
 
   /**
@@ -909,6 +918,27 @@ public enum DefaultDriverOption implements DriverOption {
    * <p>Value-type: boolean
    */
   PREPARED_CACHE_WEAK_VALUES("advanced.prepared-statements.prepared-cache.weak-values"),
+
+  /**
+   * The classes of session-wide components that track the outcome of requests.
+   *
+   * <p>Value-type: List of {@link String}
+   */
+  REQUEST_TRACKER_CLASSES("advanced.request-tracker.classes"),
+
+  /**
+   * The classes of session-wide components that listen for node state changes.
+   *
+   * <p>Value-type: List of {@link String}
+   */
+  METADATA_NODE_STATE_LISTENER_CLASSES("advanced.node-state-listener.classes"),
+
+  /**
+   * The classes of session-wide components that listen for schema changes.
+   *
+   * <p>Value-type: List of {@link String}
+   */
+  METADATA_SCHEMA_CHANGE_LISTENER_CLASSES("advanced.schema-change-listener.classes"),
   ;
 
   private final String path;

core/src/main/java/com/datastax/oss/driver/api/core/config/OngoingConfigOptions.java

@@ -19,6 +19,7 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /** An object where config options can be set programmatically. */
 public interface OngoingConfigOptions<SelfT extends OngoingConfigOptions<SelfT>> {
@@ -59,6 +60,15 @@ default SelfT withClass(@NonNull DriverOption option, @NonNull Class<?> value) {
     return withString(option, value.getName());
   }
 
+  /**
+   * Note that this is just a shortcut to call {@link #withStringList(DriverOption, List)} with
+   * class names obtained from {@link Class#getName()}.
+   */
+  @NonNull
+  default SelfT withClassList(@NonNull DriverOption option, @NonNull List<Class<?>> values) {
+    return withStringList(option, values.stream().map(Class::getName).collect(Collectors.toList()));
+  }
+
   @NonNull
   SelfT withStringList(@NonNull DriverOption option, @NonNull List<String> value);
 

core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java

@@ -280,10 +280,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) {
     map.put(TypedDriverOption.TIMESTAMP_GENERATOR_DRIFT_WARNING_THRESHOLD, Duration.ofSeconds(1));
     map.put(TypedDriverOption.TIMESTAMP_GENERATOR_DRIFT_WARNING_INTERVAL, Duration.ofSeconds(10));
     map.put(TypedDriverOption.TIMESTAMP_GENERATOR_FORCE_JAVA_CLOCK, false);
-    map.put(TypedDriverOption.REQUEST_TRACKER_CLASS, "NoopRequestTracker");
     map.put(TypedDriverOption.REQUEST_THROTTLER_CLASS, "PassThroughRequestThrottler");
-    map.put(TypedDriverOption.METADATA_NODE_STATE_LISTENER_CLASS, "NoopNodeStateListener");
-    map.put(TypedDriverOption.METADATA_SCHEMA_CHANGE_LISTENER_CLASS, "NoopSchemaChangeListener");
     map.put(TypedDriverOption.ADDRESS_TRANSLATOR_CLASS, "PassThroughAddressTranslator");
     map.put(TypedDriverOption.RESOLVE_CONTACT_POINTS, true);
     map.put(TypedDriverOption.PROTOCOL_MAX_FRAME_LENGTH, 256L * 1024 * 1024);

core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java

@@ -254,9 +254,21 @@ public String toString() {
   public static final TypedDriverOption<Duration> TIMESTAMP_GENERATOR_DRIFT_WARNING_INTERVAL =
       new TypedDriverOption<>(
           DefaultDriverOption.TIMESTAMP_GENERATOR_DRIFT_WARNING_INTERVAL, GenericType.DURATION);
-  /** The class of a session-wide component that tracks the outcome of requests. */
+
+  /**
+   * The class of a session-wide component that tracks the outcome of requests.
+   *
+   * @deprecated Use {@link #REQUEST_TRACKER_CLASSES} instead.
+   */
+  @Deprecated
   public static final TypedDriverOption<String> REQUEST_TRACKER_CLASS =
       new TypedDriverOption<>(DefaultDriverOption.REQUEST_TRACKER_CLASS, GenericType.STRING);
+
+  /** The classes of session-wide components that track the outcome of requests. */
+  public static final TypedDriverOption<List<String>> REQUEST_TRACKER_CLASSES =
+      new TypedDriverOption<>(
+          DefaultDriverOption.REQUEST_TRACKER_CLASSES, GenericType.listOf(String.class));
+
   /** Whether to log successful requests. */
   public static final TypedDriverOption<Boolean> REQUEST_LOGGER_SUCCESS_ENABLED =
       new TypedDriverOption<>(
@@ -312,14 +324,39 @@ public String toString() {
   public static final TypedDriverOption<Duration> REQUEST_THROTTLER_DRAIN_INTERVAL =
       new TypedDriverOption<>(
           DefaultDriverOption.REQUEST_THROTTLER_DRAIN_INTERVAL, GenericType.DURATION);
-  /** The class of a session-wide component that listens for node state changes. */
+
+  /**
+   * The class of a session-wide component that listens for node state changes.
+   *
+   * @deprecated Use {@link #METADATA_NODE_STATE_LISTENER_CLASSES} instead.
+   */
+  @Deprecated
   public static final TypedDriverOption<String> METADATA_NODE_STATE_LISTENER_CLASS =
       new TypedDriverOption<>(
           DefaultDriverOption.METADATA_NODE_STATE_LISTENER_CLASS, GenericType.STRING);
-  /** The class of a session-wide component that listens for schema changes. */
+
+  /**
+   * The class of a session-wide component that listens for schema changes.
+   *
+   * @deprecated Use {@link #METADATA_SCHEMA_CHANGE_LISTENER_CLASSES} instead.
+   */
+  @Deprecated
   public static final TypedDriverOption<String> METADATA_SCHEMA_CHANGE_LISTENER_CLASS =
       new TypedDriverOption<>(
           DefaultDriverOption.METADATA_SCHEMA_CHANGE_LISTENER_CLASS, GenericType.STRING);
+
+  /** The classes of session-wide components that listen for node state changes. */
+  public static final TypedDriverOption<List<String>> METADATA_NODE_STATE_LISTENER_CLASSES =
+      new TypedDriverOption<>(
+          DefaultDriverOption.METADATA_NODE_STATE_LISTENER_CLASSES,
+          GenericType.listOf(String.class));
+
+  /** The classes of session-wide components that listen for schema changes. */
+  public static final TypedDriverOption<List<String>> METADATA_SCHEMA_CHANGE_LISTENER_CLASSES =
+      new TypedDriverOption<>(
+          DefaultDriverOption.METADATA_SCHEMA_CHANGE_LISTENER_CLASSES,
+          GenericType.listOf(String.class));
+
   /**
    * The class of the address translator to use to convert the addresses sent by Cassandra nodes
    * into ones that the driver uses to connect.

core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LoadBalancingPolicy.java

@@ -19,22 +19,40 @@
 import com.datastax.oss.driver.api.core.metadata.NodeState;
 import com.datastax.oss.driver.api.core.session.Request;
 import com.datastax.oss.driver.api.core.session.Session;
+import com.datastax.oss.driver.api.core.tracker.RequestTracker;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Queue;
 import java.util.UUID;
 
 /** Decides which Cassandra nodes to contact for each query. */
 public interface LoadBalancingPolicy extends AutoCloseable {
 
+  /**
+   * Returns an optional {@link RequestTracker} to be registered with the session. Registering a
+   * request tracker allows load-balancing policies to track node latencies in order to pick the
+   * fastest ones.
+   *
+   * <p>This method is invoked only once during session configuration, and before any other methods
+   * in this interface. Note that at this point, the driver hasn't connected to any node yet.
+   *
+   * @since 4.13.0
+   */
+  @NonNull
+  default Optional<RequestTracker> getRequestTracker() {
+    return Optional.empty();
+  }
+
   /**
    * Initializes this policy with the nodes discovered during driver initialization.
    *
    * <p>This method is guaranteed to be called exactly once per instance, and before any other
-   * method in this class. At this point, the driver has successfully connected to one of the
-   * contact points, and performed a first refresh of topology information (by default, the contents
-   * of {@code system.peers}), to discover other nodes in the cluster.
+   * method in this interface except {@link #getRequestTracker()}. At this point, the driver has
+   * successfully connected to one of the contact points, and performed a first refresh of topology
+   * information (by default, the contents of {@code system.peers}), to discover other nodes in the
+   * cluster.
    *
    * <p>This method must call {@link DistanceReporter#setDistance(Node, NodeDistance)
    * distanceReporter.setDistance} for each provided node (otherwise that node will stay at distance
@@ -50,7 +68,7 @@ public interface LoadBalancingPolicy extends AutoCloseable {
    * @param nodes all the nodes that are known to exist in the cluster (regardless of their state)
    *     at the time of invocation.
    * @param distanceReporter an object that will be used by the policy to signal distance changes.
-   *     Implementations will typically store a this in a field, since new nodes may get {@link
+   *     Implementations will typically store this in a field, since new nodes may get {@link
    *     #onAdd(Node) added} later and will need to have their distance set (or the policy might
    *     change distances dynamically over time).
    */

core/src/main/java/com/datastax/oss/driver/api/core/metadata/NodeStateListener.java

@@ -23,8 +23,9 @@
 /**
  * A listener that gets notified when nodes states change.
  *
- * <p>An implementation of this interface can be registered in the configuration, or with {@link
- * SessionBuilder#withNodeStateListener(NodeStateListener)}.
+ * <p>Implementations of this interface can be registered either via the configuration (see {@code
+ * reference.conf} in the manual or core driver JAR), or programmatically via {@link
+ * SessionBuilder#addNodeStateListener(NodeStateListener)}.
  *
  * <p>Note that the methods defined by this interface will be executed by internal driver threads,
  * and are therefore expected to have short execution times. If you need to perform long
@@ -33,6 +34,9 @@
  *
  * <p>If you implement this interface but don't need to implement all the methods, extend {@link
  * NodeStateListenerBase}.
+ *
+ * <p>If your implementation of this interface requires access to a fully-initialized session,
+ * consider wrapping it in a {@link SafeInitNodeStateListener}.
  */
 public interface NodeStateListener extends AutoCloseable {
 

core/src/main/java/com/datastax/oss/driver/api/core/metadata/schema/SchemaChangeListener.java

@@ -23,8 +23,9 @@
 /**
  * Tracks schema changes.
  *
- * <p>An implementation of this interface can be registered in the configuration, or with {@link
- * SessionBuilder#withSchemaChangeListener(SchemaChangeListener)}.
+ * <p>Implementations of this interface can be registered either via the configuration (see {@code
+ * reference.conf} in the manual or core driver JAR), or programmatically via {@link
+ * SessionBuilder#addSchemaChangeListener(SchemaChangeListener)}.
  *
  * <p>Note that the methods defined by this interface will be executed by internal driver threads,
  * and are therefore expected to have short execution times. If you need to perform long

core/src/main/java/com/datastax/oss/driver/api/core/session/ProgrammaticArguments.java

@@ -25,6 +25,9 @@
 import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
 import com.datastax.oss.driver.api.core.type.codec.registry.MutableCodecRegistry;
 import com.datastax.oss.driver.internal.core.loadbalancing.helper.NodeFilterToDistanceEvaluatorAdapter;
+import com.datastax.oss.driver.internal.core.metadata.MultiplexingNodeStateListener;
+import com.datastax.oss.driver.internal.core.metadata.schema.MultiplexingSchemaChangeListener;
+import com.datastax.oss.driver.internal.core.tracker.MultiplexingRequestTracker;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -33,6 +36,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.UUID;
 import java.util.function.Predicate;
 
@@ -217,18 +221,77 @@ public Builder withNodeStateListener(@Nullable NodeStateListener nodeStateListen
       return this;
     }
 
+    @NonNull
+    public Builder addNodeStateListener(@NonNull NodeStateListener nodeStateListener) {
+      Objects.requireNonNull(nodeStateListener, "nodeStateListener cannot be null");
+      if (this.nodeStateListener == null) {
+        this.nodeStateListener = nodeStateListener;
+      } else {
+        NodeStateListener previousListener = this.nodeStateListener;
+        if (previousListener instanceof MultiplexingNodeStateListener) {
+          ((MultiplexingNodeStateListener) previousListener).register(nodeStateListener);
+        } else {
+          MultiplexingNodeStateListener multiplexingNodeStateListener =
+              new MultiplexingNodeStateListener();
+          multiplexingNodeStateListener.register(previousListener);
+          multiplexingNodeStateListener.register(nodeStateListener);
+          this.nodeStateListener = multiplexingNodeStateListener;
+        }
+      }
+      return this;
+    }
+
     @NonNull
     public Builder withSchemaChangeListener(@Nullable SchemaChangeListener schemaChangeListener) {
       this.schemaChangeListener = schemaChangeListener;
       return this;
     }
 
+    @NonNull
+    public Builder addSchemaChangeListener(@NonNull SchemaChangeListener schemaChangeListener) {
+      Objects.requireNonNull(schemaChangeListener, "schemaChangeListener cannot be null");
+      if (this.schemaChangeListener == null) {
+        this.schemaChangeListener = schemaChangeListener;
+      } else {
+        SchemaChangeListener previousListener = this.schemaChangeListener;
+        if (previousListener instanceof MultiplexingSchemaChangeListener) {
+          ((MultiplexingSchemaChangeListener) previousListener).register(schemaChangeListener);
+        } else {
+          MultiplexingSchemaChangeListener multiplexingSchemaChangeListener =
+              new MultiplexingSchemaChangeListener();
+          multiplexingSchemaChangeListener.register(previousListener);
+          multiplexingSchemaChangeListener.register(schemaChangeListener);
+          this.schemaChangeListener = multiplexingSchemaChangeListener;
+        }
+      }
+      return this;
+    }
+
     @NonNull
     public Builder withRequestTracker(@Nullable RequestTracker requestTracker) {
       this.requestTracker = requestTracker;
       return this;
     }
 
+    @NonNull
+    public Builder addRequestTracker(@NonNull RequestTracker requestTracker) {
+      Objects.requireNonNull(requestTracker, "requestTracker cannot be null");
+      if (this.requestTracker == null) {
+        this.requestTracker = requestTracker;
+      } else {
+        RequestTracker previousTracker = this.requestTracker;
+        if (previousTracker instanceof MultiplexingRequestTracker) {
+          ((MultiplexingRequestTracker) previousTracker).register(requestTracker);
+        } else {
+          MultiplexingRequestTracker multiplexingRequestTracker = new MultiplexingRequestTracker();
+          multiplexingRequestTracker.register(previousTracker);
+          multiplexingRequestTracker.register(requestTracker);
+          this.requestTracker = multiplexingRequestTracker;
+        }
+      }
+      return this;
+    }
+
     @NonNull
     public Builder withLocalDatacenter(
         @NonNull String profileName, @NonNull String localDatacenter) {