fix(#384): Ignore local racks for LWT statements in DefaultLoadBalancingPolicy.newQUeryPlan().

Author: nikagra

core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java

@@ -135,7 +135,9 @@ protected Optional<String> discoverLocalDc(@NonNull Map<UUID, Node> nodes) {
   @Override
   public Queue<Node> newQueryPlan(@Nullable Request request, @Nullable Session session) {
     List<Node> replicas = getReplicas(request, session);
-    boolean isLWT = Objects.nonNull(request) && request.getRequestType() == RequestRoutingType.LWT;
+    RequestRoutingType requestType =
+        Objects.nonNull(request) ? request.getRequestType() : RequestRoutingType.REGULAR;
+    boolean isLWT = requestType == RequestRoutingType.LWT;
     Object[] currentNodes =
         isLWT ? replicas.toArray() : getLiveNodes().dc(getLocalDatacenter()).toArray();
     if (Objects.nonNull(request)
@@ -145,12 +147,13 @@ public Queue<Node> newQueryPlan(@Nullable Request request, @Nullable Session ses
 
     int replicaCount = 0; // in currentNodes
     if (!replicas.isEmpty()) {
-      Pair<Integer, Integer> counts = moveReplicasToFront(currentNodes, replicas);
+      Pair<Integer, Integer> counts = moveReplicasToFront(requestType, currentNodes, replicas);
       replicaCount = counts.getLeft();
       int localRackReplicaCount = counts.getRight(); // in currentNodes
 
       if (replicaCount > 1) {
-        shuffleLocalRackReplicasAndReplicas(currentNodes, replicaCount, localRackReplicaCount);
+        shuffleLocalRackReplicasAndReplicas(
+            requestType, currentNodes, replicaCount, localRackReplicaCount);
 
         if (replicaCount > 2 && avoidSlowReplicas) {
           avoidSlowReplicas(Objects.requireNonNull(session), currentNodes, replicaCount);
@@ -171,13 +174,15 @@ public Queue<Node> newQueryPlan(@Nullable Request request, @Nullable Session ses
     return maybeAddDcFailover(request, plan);
   }
 
-  private Pair<Integer, Integer> moveReplicasToFront(Object[] currentNodes, Set<Node> allReplicas) {
+  private Pair<Integer, Integer> moveReplicasToFront(
+      RequestRoutingType routingType, Object[] currentNodes, List<Node> allReplicas) {
     int replicaCount = 0, localRackReplicaCount = 0;
     for (int i = 0; i < currentNodes.length; i++) {
       Node node = (Node) currentNodes[i];
       if (allReplicas.contains(node)) {
         if (Objects.equals(node.getRack(), getLocalRack())
-            && Objects.equals(node.getDatacenter(), getLocalDatacenter())) {
+            && Objects.equals(node.getDatacenter(), getLocalDatacenter())
+            && routingType != RequestRoutingType.LWT) {
           ArrayUtils.bubbleUp(currentNodes, i, localRackReplicaCount);
           localRackReplicaCount++;
         } else {
@@ -190,8 +195,13 @@ private Pair<Integer, Integer> moveReplicasToFront(Object[] currentNodes, Set<No
   }
 
   private void shuffleLocalRackReplicasAndReplicas(
-      Object[] currentNodes, int replicaCount, int localRackReplicaCount) {
-    if (getLocalRack() != null && localRackReplicaCount > 0) {
+      RequestRoutingType routingType,
+      Object[] currentNodes,
+      int replicaCount,
+      int localRackReplicaCount) {
+    if (routingType != RequestRoutingType.LWT
+        && getLocalRack() != null
+        && localRackReplicaCount > 0) {
       // Shuffle only replicas that are in the local rack
       shuffleHead(currentNodes, localRackReplicaCount);
       // Shuffles only replicas that are not in local rack