JVMCBC-1696 Should not make bucketful KV connections to nodes that aren't hosting the bucket

Motivation
----------
During the migration to the new cluster topology classes,
there was a regression that caused `core.reconfigureBuckets`
to create bucket-scoped KV endpoints for nodes that
aren't hosting the bucket.

During rebalance, this causes noise in the logs due to fruitless
connection attempts, and more importantly causes `Cluster.diagnostics()`
to report the cluster is in a degraded state because the doomed
endpoints are not connected.

Modifications
-------------
When reconfiguring a bucket, transform the nodes by removing
the KV service from any node that is not hosting the bucket.
This echoes the behavior of the old BucketConfig class.

Note that we need to process the whole node list from the
"global" part of the topology so we can remove existing
KV endpoints that are no longer required by the updated topology.

Change-Id: I8eaf79d39033d4a24cfdd5b1eb18864d296b931e
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/235828
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Michael Reiche <michael.reiche@couchbase.com>

Author: dnault

core-io/src/main/java/com/couchbase/client/core/Core.java

@@ -96,6 +96,7 @@
 import com.couchbase.client.core.topology.ClusterIdentifierUtil;
 import com.couchbase.client.core.topology.ClusterTopology;
 import com.couchbase.client.core.topology.ClusterTopologyWithBucket;
+import com.couchbase.client.core.topology.HostAndServicePorts;
 import com.couchbase.client.core.topology.NodeIdentifier;
 import com.couchbase.client.core.transaction.cleanup.CoreTransactionsCleanup;
 import com.couchbase.client.core.transaction.context.CoreTransactionsContext;
@@ -104,8 +105,8 @@
 import com.couchbase.client.core.util.Deadline;
 import com.couchbase.client.core.util.LatestStateSubscription;
 import com.couchbase.client.core.util.NanoTimestamp;
-import org.slf4j.LoggerFactory;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -135,6 +136,8 @@
 import static com.couchbase.client.core.util.ConnectionStringUtil.sanityCheckPorts;
 import static java.util.Collections.emptySet;
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
 
 /**
  * The main entry point into the core layer.
@@ -823,18 +826,35 @@ private Mono<Void> reconfigureBuckets(final Flux<ClusterTopologyWithBucket> buck
       .then();
   }
 
+  /**
+   * Returns the given topology's nodes, but with the KV service removed from nodes that aren't hosting the bucket.
+   * <p>
+   * This prevents the SDK from creating doomed KV endpoints that can never connect because
+   * "select bucket" returns an error when the node isn't hosting the bucket.
+   */
+  private static List<HostAndServicePorts> nodesWithoutIrrelevantKvServices(ClusterTopologyWithBucket topology) {
+    Set<NodeIdentifier> nodeIdsServicingBucket =
+      topology.bucket().nodes().stream()
+        .map(HostAndServicePorts::id)
+        .collect(toSet());
+
+    return topology.nodes().stream()
+      .map(node -> nodeIdsServicingBucket.contains(node.id()) ? node : node.without(ServiceType.KV))
+      .collect(toList());
+  }
+
   /**
    * @param bucketName pass non-null if using the topology to configure bucket-scoped services.
-   *
-   * @implNote Maybe in the future we can inspect the ClusterTopology to see if it has a BucketTopology,
-   * and get the bucket name from there. However, let's make it explicit for now; this leaves the door open
-   * to using a ClusterTopologyWithBucket to configure global services (by passing a null bucket name).
    */
   private Mono<Void> reconfigureGlobalOrBucket(
     ClusterTopology topology,
     @Nullable String bucketName
   ) {
-    return Flux.fromIterable(topology.nodes())
+    List<HostAndServicePorts> nodes = bucketName == null
+      ? topology.nodes()
+      : nodesWithoutIrrelevantKvServices(topology.requireBucket());
+
+    return Flux.fromIterable(nodes)
       .flatMap(ni -> {
         Flux<Void> serviceRemoveFlux = Flux
           .fromArray(ServiceType.values())

core-io/src/main/java/com/couchbase/client/core/topology/BucketTopology.java

@@ -17,7 +17,6 @@
 package com.couchbase.client.core.topology;
 
 import com.couchbase.client.core.annotation.Stability;
-import com.couchbase.client.core.config.BucketConfig;
 import com.couchbase.client.core.deps.com.fasterxml.jackson.databind.node.ObjectNode;
 import com.couchbase.client.core.node.MemcachedHashingStrategy;
 import org.jspecify.annotations.Nullable;
@@ -50,9 +49,6 @@ default boolean hasCapability(BucketCapability capability) {
   /**
    * Returns the subset of cluster nodes that are ready to service
    * requests for this bucket.
-   * <p>
-   * This method is a candidate for eventual removal, since it's used only by
-   * unit tests and the code that converts a BucketTopology to a legacy {@link BucketConfig}.
    */
   @Stability.Internal
   List<HostAndServicePorts> nodes();