JAVA-2582: Don't propagate a future into SchemaQueriesFactory (scylladb#23)

Author: tomekl007

changelog/README.md

@@ -8,6 +8,8 @@ This version brings in all functionality that was formerly only in the DataStax
 such as the built-in support for reactive programming. Going forward, all new features will be 
 implemented in this single driver.
 
+
+- [improvement] JAVA-2582: Don't propagate a future into SchemaQueriesFactory
 - [documentation] JAVA-2542: JAVA-2542: Improve the javadocs of methods in CqlSession
 - [documentation] JAVA-2609: Add docs for proxy authentication to unified driver
 - [improvement] JAVA-2554: Improve efficiency of InsightsClient by improving supportsInsights check

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/MetadataManager.java

@@ -432,9 +432,7 @@ private void startSchemaRequest(CompletableFuture<RefreshSchemaResult> refreshFu
                     refreshFuture.completeExceptionally(agreementError);
                   } else {
                     schemaQueriesFactory
-                        .newInstance(
-                            // TODO remove this unused parameter (see JAVA-2582)
-                            null)
+                        .newInstance()
                         .execute()
                         .thenApplyAsync(this::parseAndApplySchemaRows, adminExecutor)
                         .whenComplete(

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/Cassandra21SchemaQueries.java

@@ -16,22 +16,16 @@
 package com.datastax.oss.driver.internal.core.metadata.schema.queries;
 
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
-import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 import net.jcip.annotations.ThreadSafe;
 
 @ThreadSafe
 public class Cassandra21SchemaQueries extends CassandraSchemaQueries {
   public Cassandra21SchemaQueries(
-      DriverChannel channel,
-      Node node,
-      CompletableFuture<Metadata> refreshFuture,
-      DriverExecutionProfile config,
-      String logPrefix) {
-    super(channel, node, refreshFuture, config, logPrefix);
+      DriverChannel channel, Node node, DriverExecutionProfile config, String logPrefix) {
+    super(channel, node, config, logPrefix);
   }
 
   @Override

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/Cassandra22SchemaQueries.java

@@ -16,22 +16,16 @@
 package com.datastax.oss.driver.internal.core.metadata.schema.queries;
 
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
-import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 import net.jcip.annotations.ThreadSafe;
 
 @ThreadSafe
 public class Cassandra22SchemaQueries extends CassandraSchemaQueries {
   public Cassandra22SchemaQueries(
-      DriverChannel channel,
-      Node node,
-      CompletableFuture<Metadata> refreshFuture,
-      DriverExecutionProfile config,
-      String logPrefix) {
-    super(channel, node, refreshFuture, config, logPrefix);
+      DriverChannel channel, Node node, DriverExecutionProfile config, String logPrefix) {
+    super(channel, node, config, logPrefix);
   }
 
   @Override

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/Cassandra3SchemaQueries.java

@@ -16,22 +16,16 @@
 package com.datastax.oss.driver.internal.core.metadata.schema.queries;
 
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
-import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 import net.jcip.annotations.ThreadSafe;
 
 @ThreadSafe
 public class Cassandra3SchemaQueries extends CassandraSchemaQueries {
   public Cassandra3SchemaQueries(
-      DriverChannel channel,
-      Node node,
-      CompletableFuture<Metadata> refreshFuture,
-      DriverExecutionProfile config,
-      String logPrefix) {
-    super(channel, node, refreshFuture, config, logPrefix);
+      DriverChannel channel, Node node, DriverExecutionProfile config, String logPrefix) {
+    super(channel, node, config, logPrefix);
   }
 
   @Override

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/Cassandra4SchemaQueries.java

@@ -16,22 +16,16 @@
 package com.datastax.oss.driver.internal.core.metadata.schema.queries;
 
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
-import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 import net.jcip.annotations.ThreadSafe;
 
 @ThreadSafe
 public class Cassandra4SchemaQueries extends Cassandra3SchemaQueries {
   public Cassandra4SchemaQueries(
-      DriverChannel channel,
-      Node node,
-      CompletableFuture<Metadata> refreshFuture,
-      DriverExecutionProfile config,
-      String logPrefix) {
-    super(channel, node, refreshFuture, config, logPrefix);
+      DriverChannel channel, Node node, DriverExecutionProfile config, String logPrefix) {
+    super(channel, node, config, logPrefix);
   }
 
   @Override

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/CassandraSchemaQueries.java

@@ -17,7 +17,6 @@
 
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
-import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminRequestHandler;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminResult;
@@ -52,25 +51,17 @@ public abstract class CassandraSchemaQueries implements SchemaQueries {
   private final String whereClause;
   // The future we return from execute, completes when all the queries are done.
   private final CompletableFuture<SchemaRows> schemaRowsFuture = new CompletableFuture<>();
-  // A future that completes later, when the whole refresh is done. We just store it here to pass it
-  // down to the next step.
-  public final CompletableFuture<Metadata> refreshFuture;
   private final long startTimeNs = System.nanoTime();
 
   // All non-final fields are accessed exclusively on adminExecutor
   private CassandraSchemaRows.Builder schemaRowsBuilder;
   private int pendingQueries;
 
   protected CassandraSchemaQueries(
-      DriverChannel channel,
-      Node node,
-      CompletableFuture<Metadata> refreshFuture,
-      DriverExecutionProfile config,
-      String logPrefix) {
+      DriverChannel channel, Node node, DriverExecutionProfile config, String logPrefix) {
     this.channel = channel;
     this.adminExecutor = channel.eventLoop();
     this.node = node;
-    this.refreshFuture = refreshFuture;
     this.logPrefix = logPrefix;
     this.timeout = config.getDuration(DefaultDriverOption.METADATA_SCHEMA_REQUEST_TIMEOUT);
     this.pageSize = config.getInt(DefaultDriverOption.METADATA_SCHEMA_REQUEST_PAGE_SIZE);
@@ -130,7 +121,7 @@ public CompletionStage<SchemaRows> execute() {
   private void executeOnAdminExecutor() {
     assert adminExecutor.inEventLoop();
 
-    schemaRowsBuilder = new CassandraSchemaRows.Builder(node, refreshFuture, logPrefix);
+    schemaRowsBuilder = new CassandraSchemaRows.Builder(node, logPrefix);
 
     query(selectKeyspacesQuery() + whereClause, schemaRowsBuilder::withKeyspaces);
     query(selectTypesQuery() + whereClause, schemaRowsBuilder::withTypes);

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/CassandraSchemaRows.java

@@ -18,7 +18,6 @@
 import com.datastax.dse.driver.api.core.metadata.DseNodeProperties;
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.Version;
-import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.adminrequest.AdminRow;
 import com.datastax.oss.driver.internal.core.metadata.schema.parsing.DataTypeClassNameParser;
@@ -33,7 +32,6 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
 import net.jcip.annotations.Immutable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -43,7 +41,6 @@ public class CassandraSchemaRows implements SchemaRows {
 
   private final Node node;
   private final DataTypeParser dataTypeParser;
-  private final CompletableFuture<Metadata> refreshFuture;
   private final List<AdminRow> keyspaces;
   private final List<AdminRow> virtualKeyspaces;
   private final Multimap<CqlIdentifier, AdminRow> tables;
@@ -58,7 +55,6 @@ public class CassandraSchemaRows implements SchemaRows {
 
   private CassandraSchemaRows(
       Node node,
-      CompletableFuture<Metadata> refreshFuture,
       DataTypeParser dataTypeParser,
       List<AdminRow> keyspaces,
       List<AdminRow> virtualKeyspaces,
@@ -73,7 +69,6 @@ private CassandraSchemaRows(
       Multimap<CqlIdentifier, AdminRow> aggregates) {
     this.node = node;
     this.dataTypeParser = dataTypeParser;
-    this.refreshFuture = refreshFuture;
     this.keyspaces = keyspaces;
     this.virtualKeyspaces = virtualKeyspaces;
     this.tables = tables;
@@ -98,11 +93,6 @@ public DataTypeParser dataTypeParser() {
     return dataTypeParser;
   }
 
-  @Override
-  public CompletableFuture<Metadata> refreshFuture() {
-    return refreshFuture;
-  }
-
   @Override
   public List<AdminRow> keyspaces() {
     return keyspaces;
@@ -162,7 +152,6 @@ public static class Builder {
     private static final Logger LOG = LoggerFactory.getLogger(Builder.class);
 
     private final Node node;
-    private final CompletableFuture<Metadata> refreshFuture;
     private final DataTypeParser dataTypeParser;
     private final String tableNameColumn;
     private final String logPrefix;
@@ -187,9 +176,8 @@ public static class Builder {
     private final Map<CqlIdentifier, ImmutableMultimap.Builder<CqlIdentifier, AdminRow>>
         indexesBuilders = new LinkedHashMap<>();
 
-    public Builder(Node node, CompletableFuture<Metadata> refreshFuture, String logPrefix) {
+    public Builder(Node node, String logPrefix) {
       this.node = node;
-      this.refreshFuture = refreshFuture;
       this.logPrefix = logPrefix;
       if (isCassandraV3OrAbove(node)) {
         this.tableNameColumn = "table_name";
@@ -323,7 +311,6 @@ private void putByKeyspaceAndTable(
     public CassandraSchemaRows build() {
       return new CassandraSchemaRows(
           node,
-          refreshFuture,
           dataTypeParser,
           keyspacesBuilder.build(),
           virtualKeyspacesBuilder.build(),

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/DefaultSchemaQueriesFactory.java

@@ -18,11 +18,9 @@
 import com.datastax.dse.driver.api.core.metadata.DseNodeProperties;
 import com.datastax.oss.driver.api.core.Version;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
-import com.datastax.oss.driver.api.core.metadata.Metadata;
 import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.channel.DriverChannel;
 import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
-import java.util.concurrent.CompletableFuture;
 import net.jcip.annotations.ThreadSafe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,7 +39,7 @@ public DefaultSchemaQueriesFactory(InternalDriverContext context) {
   }
 
   @Override
-  public SchemaQueries newInstance(CompletableFuture<Metadata> refreshFuture) {
+  public SchemaQueries newInstance() {
     DriverChannel channel = context.getControlConnection().channel();
     if (channel == null || channel.closeFuture().isDone()) {
       throw new IllegalStateException("Control channel not available, aborting schema refresh");
@@ -57,11 +55,10 @@ public SchemaQueries newInstance(CompletableFuture<Metadata> refreshFuture) {
                         "Could not find control node metadata "
                             + channel.getEndPoint()
                             + ", aborting schema refresh"));
-    return newInstance(node, channel, refreshFuture);
+    return newInstance(node, channel);
   }
 
-  protected SchemaQueries newInstance(
-      Node node, DriverChannel channel, CompletableFuture<Metadata> refreshFuture) {
+  protected SchemaQueries newInstance(Node node, DriverChannel channel) {
 
     DriverExecutionProfile config = context.getConfig().getDefaultProfile();
 
@@ -73,13 +70,13 @@ protected SchemaQueries newInstance(
           "[{}] Sending schema queries to {} with DSE version {}", logPrefix, node, dseVersion);
       // 4.8 is the oldest version supported, which uses C* 2.1 schema
       if (dseVersion.compareTo(Version.V5_0_0) < 0) {
-        return new Cassandra21SchemaQueries(channel, node, refreshFuture, config, logPrefix);
+        return new Cassandra21SchemaQueries(channel, node, config, logPrefix);
       } else if (dseVersion.compareTo(Version.V6_7_0) < 0) {
         // 5.0 - 6.7 uses C* 3.0 schema
-        return new Cassandra3SchemaQueries(channel, node, refreshFuture, config, logPrefix);
+        return new Cassandra3SchemaQueries(channel, node, config, logPrefix);
       } else {
         // 6.7+ uses C* 4.0 schema
-        return new Cassandra4SchemaQueries(channel, node, refreshFuture, config, logPrefix);
+        return new Cassandra4SchemaQueries(channel, node, config, logPrefix);
       }
     } else {
       Version cassandraVersion = node.getCassandraVersion();
@@ -96,13 +93,13 @@ protected SchemaQueries newInstance(
       LOG.debug(
           "[{}] Sending schema queries to {} with version {}", logPrefix, node, cassandraVersion);
       if (cassandraVersion.compareTo(Version.V2_2_0) < 0) {
-        return new Cassandra21SchemaQueries(channel, node, refreshFuture, config, logPrefix);
+        return new Cassandra21SchemaQueries(channel, node, config, logPrefix);
       } else if (cassandraVersion.compareTo(Version.V3_0_0) < 0) {
-        return new Cassandra22SchemaQueries(channel, node, refreshFuture, config, logPrefix);
+        return new Cassandra22SchemaQueries(channel, node, config, logPrefix);
       } else if (cassandraVersion.compareTo(Version.V4_0_0) < 0) {
-        return new Cassandra3SchemaQueries(channel, node, refreshFuture, config, logPrefix);
+        return new Cassandra3SchemaQueries(channel, node, config, logPrefix);
       } else {
-        return new Cassandra4SchemaQueries(channel, node, refreshFuture, config, logPrefix);
+        return new Cassandra4SchemaQueries(channel, node, config, logPrefix);
       }
     }
   }

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/schema/queries/SchemaQueriesFactory.java

@@ -15,9 +15,6 @@
  */
 package com.datastax.oss.driver.internal.core.metadata.schema.queries;
 
-import com.datastax.oss.driver.api.core.metadata.Metadata;
-import java.util.concurrent.CompletableFuture;
-
 public interface SchemaQueriesFactory {
-  SchemaQueries newInstance(CompletableFuture<Metadata> refreshFuture);
+  SchemaQueries newInstance();
 }