fix: resolve merge conflicts from apache/cassandra-java-driver 4.19.2

- BasicLoadBalancingPolicy: add missing closing brace for getLocalRack()
- DefaultLoadBalancingPolicy: remove unused ThreadLocalRandom import
- OptionalLocalDcHelper: add missing LinkedHashSet import; restore correct
  checkLocalDatacenterCompatibility implementation from upstream
- InitialNodeListRefresh: add missing Set import; add contactPoints field
  and constructor arg; restore upstream compute() logic
- InitialNodeListRefreshTest: replace bad merge artifact with upstream
  4.19.2 version (contactPoint1/contactPoint2 setup + ImmutableSet args)
- CqlRequestHandler: rename logPrefix -> handlerLogPrefix in sendRequest()
  calls to session.getChannel()
- manual/core/README.md: add request_id/* to toctree

Author: nikagra

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

@@ -1154,7 +1154,7 @@ public enum DefaultDriverOption implements DriverOption {
    * }
    * </pre>
    *
-   * Note: subnets must be represented as prefix blocks, see {@link
+   * Note: subnets must be represented as prefix blocks, see {@code
    * inet.ipaddr.Address#isPrefixBlock()}.
    *
    * <p>Value type: {@link java.util.Map Map}&#60;{@link String},{@link String}&#62;

core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareAsyncProcessor.java

@@ -162,6 +162,11 @@ public CompletionStage<PreparedStatement> process(
                   });
         }
       }
+      // If the future is already completed, return it directly to maintain a strong reference
+      // in the cache and avoid premature GC with weakValues() (ScyllaDB PR #892).
+      if (result.isDone()) {
+        return result;
+      }
       // Return a defensive copy. So if a client cancels its request, the cache won't be impacted
       // nor a potential concurrent request.
       return result.thenApply(x -> x); // copy() is available only since Java 9

core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java

@@ -384,15 +384,15 @@ private void sendRequest(
         || (channel =
                 session.getChannel(
                     node,
-                    logPrefix,
+                    handlerLogPrefix,
                     getRoutingToken(statement),
                     getShardFromTabletMap(statement, node, getRoutingToken(statement))))
             == null) {
       while (!result.isDone() && (node = queryPlan.poll()) != null) {
         channel =
             session.getChannel(
                 node,
-                logPrefix,
+                handlerLogPrefix,
                 getRoutingToken(statement),
                 getShardFromTabletMap(statement, node, getRoutingToken(statement)));
         if (channel != null) {
@@ -672,7 +672,7 @@ private NodeResponseCallback(
       this.execution = execution;
       this.retryCount = retryCount;
       this.scheduleNextExecution = scheduleNextExecution;
-      this.logPrefix = logPrefix + "|" + execution;
+      this.logPrefix = logPrefix;
     }
 
     // this gets invoked once the write completes.

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

@@ -174,6 +174,9 @@ public BasicLoadBalancingPolicy(@NonNull DriverContext context, @NonNull String
   private RequestRoutingMethod parseLwtRequestRoutingMethod() {
     String methodString =
         profile.getString(DefaultDriverOption.LOAD_BALANCING_DEFAULT_LWT_REQUEST_ROUTING_METHOD);
+    if (methodString == null) {
+      return RequestRoutingMethod.PRESERVE_REPLICA_ORDER;
+    }
     try {
       return RequestRoutingMethod.valueOf(methodString.toUpperCase());
     } catch (IllegalArgumentException e) {
@@ -240,6 +243,7 @@ public String getLocalDatacenter() {
   @Nullable
   protected String getLocalRack() {
     return localRack;
+  }
 
   @NonNull
   @Override

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

@@ -48,7 +48,6 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicLongArray;
 import net.jcip.annotations.ThreadSafe;
 import org.slf4j.Logger;

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

@@ -26,6 +26,7 @@
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -71,6 +72,24 @@ public Optional<String> discoverLocalDc(@NonNull Map<UUID, Node> nodes) {
     if (localDc.isPresent()) {
       checkLocalDatacenterCompatibility(
           localDc.get(), context.getMetadataManager().getContactPoints());
+      // Also warn if the configured DC doesn't match any node in the cluster
+      if (!nodes.isEmpty()) {
+        boolean found = false;
+        for (Node node : nodes.values()) {
+          if (localDc.get().equals(node.getDatacenter())) {
+            found = true;
+            break;
+          }
+        }
+        if (!found) {
+          LOG.warn(
+              "[{}] Configured local DC '{}' does not match any node's datacenter"
+                  + " (available DCs: {}); please verify your configuration",
+              logPrefix,
+              localDc.get(),
+              formatDcs(nodes.values()));
+        }
+      }
     } else {
       LOG.debug("[{}] Local DC not set, DC awareness will be disabled", logPrefix);
     }
@@ -96,16 +115,15 @@ protected void checkLocalDatacenterCompatibility(
           badContactPoints.add(node);
         }
       }
-      if (!found) {
+      if (!badContactPoints.isEmpty()) {
         LOG.warn(
-            "[{}] Configured local DC '{}' does not match any node's datacenter"
-                + " (available DCs: {}); please verify your configuration",
+            "[{}] You specified {} as the local DC, but some contact points are from a different DC: {}; "
+                + "please provide the correct local DC, or check your contact points",
             logPrefix,
             localDc,
-            formatDcs(nodes.values()));
+            formatNodesAndDcs(badContactPoints));
       }
     }
-    return Optional.of(localDc);
   }
 
   /**

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

@@ -17,17 +17,14 @@
  */
 package com.datastax.oss.driver.internal.core.metadata;
 
-import com.datastax.oss.driver.api.core.metadata.EndPoint;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
 import com.datastax.oss.driver.internal.core.metadata.token.TokenFactory;
 import com.datastax.oss.driver.internal.core.metadata.token.TokenFactoryRegistry;
 import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
-import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import net.jcip.annotations.ThreadSafe;
@@ -65,10 +62,7 @@ public Result compute(
     }
 
     Map<UUID, DefaultNode> newNodes = new HashMap<>();
-    // Contact point nodes don't have host ID as well as other info yet, so we fill them with node
-    // info found on first match by endpoint
-    Set<EndPoint> matchedContactPoints = new HashSet<>();
-    List<DefaultNode> addedNodes = new ArrayList<>();
+    ImmutableList.Builder<Object> eventsBuilder = ImmutableList.builder();
 
     for (NodeInfo nodeInfo : nodeInfos) {
       UUID hostId = nodeInfo.getHostId();
@@ -80,17 +74,15 @@ public Result compute(
             hostId,
             newNodes.get(hostId));
       } else {
-        EndPoint endPoint = nodeInfo.getEndPoint();
-        DefaultNode contactPointNode = findContactPointNode(endPoint);
         DefaultNode node;
-        if (contactPointNode == null || matchedContactPoints.contains(endPoint)) {
-          node = new DefaultNode(endPoint, context);
-          addedNodes.add(node);
-          LOG.debug("[{}] Adding new node {}", logPrefix, node);
+        DefaultNode existing = existingByHostId.get(hostId);
+        if (existing != null) {
+          node = existing;
+          LOG.debug("[{}] Reusing existing node {}", logPrefix, node);
         } else {
-          matchedContactPoints.add(contactPointNode.getEndPoint());
-          node = contactPointNode;
-          LOG.debug("[{}] Copying contact point {}", logPrefix, node);
+          node = new DefaultNode(nodeInfo.getEndPoint(), context);
+          LOG.debug("[{}] Adding new node {}", logPrefix, node);
+          eventsBuilder.add(NodeStateEvent.added(node));
         }
         if (tokenMapEnabled && tokenFactory == null && nodeInfo.getPartitioner() != null) {
           tokenFactory = tokenFactoryRegistry.tokenFactoryFor(nodeInfo.getPartitioner());
@@ -100,13 +92,13 @@ public Result compute(
       }
     }
 
-    ImmutableList.Builder<Object> eventsBuilder = ImmutableList.builder();
-    for (DefaultNode addedNode : addedNodes) {
-      eventsBuilder.add(NodeStateEvent.added(addedNode));
-    }
-    for (DefaultNode contactPoint : contactPoints) {
-      if (!matchedContactPoints.contains(contactPoint.getEndPoint())) {
-        eventsBuilder.add(NodeStateEvent.removed(contactPoint));
+    for (Map.Entry<UUID, DefaultNode> entry : existingByHostId.entrySet()) {
+      if (!newNodes.containsKey(entry.getKey())) {
+        LOG.warn(
+            "[{}] Pre-registered node {} was not found in the node list refresh",
+            logPrefix,
+            entry.getValue());
+        eventsBuilder.add(NodeStateEvent.removed(entry.getValue()));
       }
     }
 
@@ -115,13 +107,4 @@ public Result compute(
             ImmutableMap.copyOf(newNodes), tokenMapEnabled, true, tokenFactory, context),
         eventsBuilder.build());
   }
-
-  private DefaultNode findContactPointNode(EndPoint endPoint) {
-    for (DefaultNode node : contactPoints) {
-      if (node.getEndPoint().equals(endPoint)) {
-        return node;
-      }
-    }
-    return null;
-  }
 }

core/src/test/java/com/datastax/oss/driver/internal/core/context/MockedDriverContextFactory.java

@@ -119,6 +119,9 @@ public static DriverExecutionProfile createProfile(String name, String localDc)
         .thenReturn("DefaultMetricsFactory");
     when(defaultProfile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER))
         .thenReturn(localDc);
+    when(defaultProfile.getString(
+            DefaultDriverOption.LOAD_BALANCING_DEFAULT_LWT_REQUEST_ROUTING_METHOD))
+        .thenReturn("PRESERVE_REPLICA_ORDER");
     return defaultProfile;
   }
 

core/src/test/java/com/datastax/oss/driver/internal/core/metadata/InitialNodeListRefreshTest.java

@@ -69,7 +69,7 @@ public void setup() {
   }
 
   @Test
-  public void should_copy_contact_points_on_first_endpoint_match_only() {
+  public void should_create_new_nodes_for_all_endpoints() {
     // Given
     Iterable<NodeInfo> newInfos =
         ImmutableList.of(
@@ -81,16 +81,6 @@ public void should_copy_contact_points_on_first_endpoint_match_only() {
                 .withEndPoint(endPoint2)
                 .withHostId(hostId4)
                 .build(),
-            DefaultNodeInfo.builder()
-                .withEndPoint(contactPoint2.getEndPoint())
-                .withHostId(hostId2)
-                .build(),
-            DefaultNodeInfo.builder().withEndPoint(endPoint3).withHostId(hostId3).build(),
-            DefaultNodeInfo.builder()
-                // address translator can translate node addresses to the same endpoints
-                .withEndPoint(contactPoint2.getEndPoint())
-                .withHostId(hostId4)
-                .build(),
             DefaultNodeInfo.builder()
                 // address translator can translate node addresses to the same endpoints
                 .withEndPoint(endPoint3)
@@ -104,22 +94,20 @@ public void should_copy_contact_points_on_first_endpoint_match_only() {
     // Then
     Map<UUID, Node> newNodes = result.newMetadata.getNodes();
     assertThat(newNodes).containsOnlyKeys(hostId1, hostId2, hostId3, hostId4, hostId5);
-    assertThat(newNodes.get(hostId1)).isEqualTo(contactPoint1);
-    assertThat(contactPoint1.getHostId()).isEqualTo(hostId1);
-    assertThat(newNodes.get(hostId2)).isEqualTo(contactPoint2);
-    assertThat(contactPoint2.getHostId()).isEqualTo(hostId2);
-    // And
-    // node has been added for the new endpoint
+    assertThat(newNodes.get(hostId1).getEndPoint()).isEqualTo(endPoint1);
+    assertThat(newNodes.get(hostId1).getHostId()).isEqualTo(hostId1);
+    assertThat(newNodes.get(hostId2).getEndPoint()).isEqualTo(endPoint2);
+    assertThat(newNodes.get(hostId2).getHostId()).isEqualTo(hostId2);
     assertThat(newNodes.get(hostId3).getEndPoint()).isEqualTo(endPoint3);
     assertThat(newNodes.get(hostId3).getHostId()).isEqualTo(hostId3);
-    // And
-    // nodes have been added for duplicated endpoints
-    assertThat(newNodes.get(hostId4).getEndPoint()).isEqualTo(contactPoint2.getEndPoint());
+    assertThat(newNodes.get(hostId4).getEndPoint()).isEqualTo(endPoint2);
     assertThat(newNodes.get(hostId4).getHostId()).isEqualTo(hostId4);
     assertThat(newNodes.get(hostId5).getEndPoint()).isEqualTo(endPoint3);
     assertThat(newNodes.get(hostId5).getHostId()).isEqualTo(hostId5);
     assertThat(result.events)
         .containsExactlyInAnyOrder(
+            NodeStateEvent.added((DefaultNode) newNodes.get(hostId1)),
+            NodeStateEvent.added((DefaultNode) newNodes.get(hostId2)),
             NodeStateEvent.added((DefaultNode) newNodes.get(hostId3)),
             NodeStateEvent.added((DefaultNode) newNodes.get(hostId4)),
             NodeStateEvent.added((DefaultNode) newNodes.get(hostId5)));

integration-tests/src/test/java/com/datastax/oss/driver/core/tracker/RequestIdGeneratorIT.java

@@ -29,24 +29,32 @@
 import com.datastax.oss.driver.api.core.session.Request;
 import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator;
 import com.datastax.oss.driver.api.testinfra.ccm.CcmRule;
+import com.datastax.oss.driver.api.testinfra.requirement.BackendRequirement;
+import com.datastax.oss.driver.api.testinfra.requirement.BackendType;
 import com.datastax.oss.driver.api.testinfra.session.SessionUtils;
 import com.datastax.oss.driver.categories.ParallelizableTests;
 import com.datastax.oss.protocol.internal.util.collection.NullAllowingImmutableMap;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
-import org.junit.Rule;
+import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import org.junit.rules.RuleChain;
 import org.junit.rules.TestRule;
 
+// Scylla does not support custom payload (RequestIdGenerator writes request IDs into the custom
+// payload); Scylla returns a truncated-frame protocol error when custom payload entries are
+// present.
+@BackendRequirement(
+    type = BackendType.CASSANDRA,
+    description = "Custom payload (request ID) not supported on Scylla")
 @Category(ParallelizableTests.class)
 public class RequestIdGeneratorIT {
-  private CcmRule ccmRule = CcmRule.getInstance();
+  private static CcmRule ccmRule = CcmRule.getInstance();
 
-  @Rule public TestRule chain = RuleChain.outerRule(ccmRule);
+  @ClassRule public static TestRule chain = RuleChain.outerRule(ccmRule);
 
   @Test
   public void should_write_uuid_to_custom_payload_with_key() {