refactor: address dkropachev review comments on PR #891

- Remove Statement.hasSerialConsistency() (too specific for public API);
  inline the serial-CL detection in each policy
- Revert LOCAL_SERIAL local-only filtering in PreserveReplicaOrderIterator;
  remote replicas now appear in the tail for both SERIAL and LOCAL_SERIAL
  (alignment with other drivers deferred to a separate task per reviewer)
- Store ConsistencyLevel (not QueryOptions) in LatencyAwarePolicy;
  expose via private isEffectiveConsistencySerial() helper
- Update TokenAwarePolicyTest: replace two separate LOCAL_SERIAL/SERIAL
  remote-replica tests with a single parameterised test covering both CLs

Author: nikagra

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

@@ -91,27 +91,6 @@ public boolean isLWT() {
     return false;
   }
 
-  /**
-   * Returns {@code true} if this statement's effective consistency level is serial ({@link
-   * ConsistencyLevel#SERIAL} or {@link ConsistencyLevel#LOCAL_SERIAL}). When no consistency level
-   * has been explicitly set on this statement, falls back to the provided default.
-   *
-   * <p>This is used by load-balancing policies to route serial-consistency statements through the
-   * LWT path, even when {@link #isLWT()} returns {@code false}.
-   *
-   * @param defaultConsistencyLevel the cluster-wide default consistency level to fall back to when
-   *     this statement has no explicit consistency level set.
-   * @return whether the effective consistency level is serial.
-   */
-  public static boolean hasSerialConsistency(
-      Statement statement, ConsistencyLevel defaultConsistencyLevel) {
-    ConsistencyLevel cl = statement.getConsistencyLevel();
-    if (cl == null) {
-      cl = defaultConsistencyLevel;
-    }
-    return cl != null && cl.isSerial();
-  }
-
   /**
    * Sets the consistency level for the query.
    *

driver-core/src/main/java/com/datastax/driver/core/policies/LatencyAwarePolicy.java

@@ -17,12 +17,12 @@
 
 import com.codahale.metrics.Gauge;
 import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
 import com.datastax.driver.core.Host;
 import com.datastax.driver.core.HostDistance;
 import com.datastax.driver.core.LatencyTracker;
 import com.datastax.driver.core.Metrics;
 import com.datastax.driver.core.MetricsUtil;
-import com.datastax.driver.core.QueryOptions;
 import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.exceptions.BootstrappingException;
 import com.datastax.driver.core.exceptions.DriverException;
@@ -103,7 +103,7 @@ public class LatencyAwarePolicy implements ChainableLoadBalancingPolicy {
   private final long retryPeriod;
   private final long minMeasure;
   private volatile Metrics metrics;
-  private volatile QueryOptions queryOptions;
+  private volatile ConsistencyLevel defaultConsistencyLevel;
 
   private LatencyAwarePolicy(
       LoadBalancingPolicy childPolicy,
@@ -220,7 +220,7 @@ public void init(Cluster cluster, Collection<Host> hosts) {
     }
     cluster.register(latencyTracker);
     metrics = cluster.getMetrics();
-    queryOptions = cluster.getConfiguration().getQueryOptions();
+    defaultConsistencyLevel = cluster.getConfiguration().getQueryOptions().getConsistencyLevel();
     if (metrics != null) {
       metrics
           .getRegistry()
@@ -265,9 +265,7 @@ public Iterator<Host> newQueryPlan(String loggedKeyspace, Statement statement) {
     // LWT routing can rely on deterministic replica ordering (e.g. by TokenAwarePolicy), and
     // latency-based reordering can undermine those assumptions.
     if (statement != null
-        && (statement.isLWT()
-            || Statement.hasSerialConsistency(
-                statement, queryOptions != null ? queryOptions.getConsistencyLevel() : null))) {
+        && (statement.isLWT() || isEffectiveConsistencySerial(statement.getConsistencyLevel()))) {
       return childPolicy.newQueryPlan(loggedKeyspace, statement);
     }
 
@@ -340,6 +338,11 @@ protected Host computeNext() {
     };
   }
 
+  private boolean isEffectiveConsistencySerial(ConsistencyLevel statementCl) {
+    ConsistencyLevel cl = statementCl != null ? statementCl : defaultConsistencyLevel;
+    return cl != null && cl.isSerial();
+  }
+
   /**
    * Returns a snapshot of the scores (latency averages) maintained by this policy.
    *

driver-core/src/main/java/com/datastax/driver/core/policies/RackAwareRoundRobinPolicy.java

@@ -250,9 +250,12 @@ public Iterator<Host> newQueryPlan(String loggedKeyspace, final Statement statem
     // hosts equally
     ConsistencyLevel defaultCl =
         configuration != null ? configuration.getQueryOptions().getConsistencyLevel() : null;
+    ConsistencyLevel effectiveCl =
+        statement != null && statement.getConsistencyLevel() != null
+            ? statement.getConsistencyLevel()
+            : defaultCl;
     final boolean isLWT =
-        statement != null
-            && (statement.isLWT() || Statement.hasSerialConsistency(statement, defaultCl));
+        statement != null && (statement.isLWT() || (effectiveCl != null && effectiveCl.isSerial()));
 
     // For LWT queries, include all local DC hosts in the first part of the plan, not just those in
     // the local rack

driver-core/src/main/java/com/datastax/driver/core/policies/TokenAwarePolicy.java

@@ -267,7 +267,6 @@ private class PreserveReplicaOrderIterator extends AbstractIterator<Host> {
     private final List<Host> replicas;
     private final String keyspace;
     private final Statement statement;
-    private final boolean localOnly;
     private List<Host> nonLocalReplicas;
     private Iterator<Host> nonLocalReplicasIterator;
     private Set<Host> returnedHosts;
@@ -278,11 +277,6 @@ public PreserveReplicaOrderIterator(String keyspace, Statement statement, List<H
       this.statement = statement;
       this.replicas = replicas;
       this.replicasIterator = replicas.iterator();
-      ConsistencyLevel cl = statement.getConsistencyLevel();
-      if (cl == null && queryOptions != null) {
-        cl = queryOptions.getConsistencyLevel();
-      }
-      this.localOnly = cl == ConsistencyLevel.LOCAL_SERIAL;
     }
 
     @Override
@@ -314,7 +308,7 @@ protected Host computeNext() {
       }
 
       // Second pass: return remote replicas that are UP and not IGNORED
-      if (nonLocalReplicas != null && !localOnly) {
+      if (nonLocalReplicas != null) {
         if (nonLocalReplicasIterator == null) {
           nonLocalReplicasIterator = nonLocalReplicas.iterator();
         }
@@ -479,8 +473,11 @@ private QueryOptions.RequestRoutingMethod getRequestRouting(Statement statement)
     if (defaultLwtRequestRoutingMethod == null) {
       return QueryOptions.RequestRoutingMethod.REGULAR;
     }
-    if (statement.isLWT()
-        || Statement.hasSerialConsistency(statement, queryOptions.getConsistencyLevel())) {
+    ConsistencyLevel cl = statement.getConsistencyLevel();
+    if (cl == null) {
+      cl = queryOptions.getConsistencyLevel();
+    }
+    if (statement.isLWT() || (cl != null && cl.isSerial())) {
       return defaultLwtRequestRoutingMethod;
     }
     return QueryOptions.RequestRoutingMethod.REGULAR;

driver-core/src/test/java/com/datastax/driver/core/policies/TokenAwarePolicyTest.java

@@ -125,6 +125,14 @@ public Object[][] shuffleProvider() {
     };
   }
 
+  @DataProvider(name = "serialConsistencyProvider")
+  public Object[][] serialConsistencyProvider() {
+    return new Object[][] {
+      {com.datastax.driver.core.ConsistencyLevel.SERIAL},
+      {com.datastax.driver.core.ConsistencyLevel.LOCAL_SERIAL}
+    };
+  }
+
   @Test(groups = "unit")
   public void should_respect_topological_order() {
     // given
@@ -737,41 +745,13 @@ public void should_not_route_serial_consistency_level_option_as_lwt() {
     assertThat(queryPlan).containsExactly(host1, host2, host4, host3);
   }
 
-  @Test(groups = "unit")
-  public void should_not_include_remote_replicas_for_local_serial() {
-    // given: a LOCAL_SERIAL statement with some replicas in remote DC
-    Statement localSerialStatement = mock(Statement.class);
-    when(localSerialStatement.isLWT()).thenReturn(false);
-    when(localSerialStatement.getConsistencyLevel())
-        .thenReturn(com.datastax.driver.core.ConsistencyLevel.LOCAL_SERIAL);
-    when(localSerialStatement.getRoutingKey(any(ProtocolVersion.class), any(CodecRegistry.class)))
-        .thenReturn(routingKey);
-    when(localSerialStatement.getKeyspace()).thenReturn(KEYSPACE);
-    when(metadata.getReplicasList(Metadata.quote(KEYSPACE), null, null, routingKey))
-        .thenReturn(Lists.newArrayList(host1, host2, host3));
-    when(childPolicy.distance(host1)).thenReturn(HostDistance.LOCAL);
-    when(childPolicy.distance(host2)).thenReturn(HostDistance.REMOTE);
-    when(childPolicy.distance(host3)).thenReturn(HostDistance.LOCAL);
-    when(childPolicy.newQueryPlan(KEYSPACE, localSerialStatement))
-        .thenReturn(Lists.newArrayList(host4, host5).iterator());
-
-    TokenAwarePolicy policy = new TokenAwarePolicy(childPolicy, TOPOLOGICAL);
-    policy.init(cluster, null);
-
-    // when
-    Iterator<Host> queryPlan = policy.newQueryPlan(KEYSPACE, localSerialStatement);
-
-    // then: only local replicas (host1, host3), no remote replica (host2), then non-replicas
-    assertThat(queryPlan).containsExactly(host1, host3, host4, host5);
-  }
-
-  @Test(groups = "unit")
-  public void should_include_remote_replicas_for_serial() {
-    // given: a SERIAL statement with some replicas in remote DC
+  @Test(groups = "unit", dataProvider = "serialConsistencyProvider")
+  public void should_include_remote_replicas_for_serial_consistency(
+      com.datastax.driver.core.ConsistencyLevel serialCl) {
+    // given: a serial-CL statement with some replicas in remote DC
     Statement serialStatement = mock(Statement.class);
     when(serialStatement.isLWT()).thenReturn(false);
-    when(serialStatement.getConsistencyLevel())
-        .thenReturn(com.datastax.driver.core.ConsistencyLevel.SERIAL);
+    when(serialStatement.getConsistencyLevel()).thenReturn(serialCl);
     when(serialStatement.getRoutingKey(any(ProtocolVersion.class), any(CodecRegistry.class)))
         .thenReturn(routingKey);
     when(serialStatement.getKeyspace()).thenReturn(KEYSPACE);
@@ -789,7 +769,7 @@ public void should_include_remote_replicas_for_serial() {
     // when
     Iterator<Host> queryPlan = policy.newQueryPlan(KEYSPACE, serialStatement);
 
-    // then: local replicas (host1, host3), then remote replica (host2), then non-replicas
+    // then: local replicas first (host1, host3), then remote replica (host2), then non-replicas
     assertThat(queryPlan).containsExactly(host1, host3, host2, host4, host5);
   }