JVMCBC-1693 Debounce GetConfig requests

Motivation
----------
During rebalance, the SDK receives many cluster topology
change notifications. Instead of fetching the new topology
once for each notification, the SDK should reduce the number
of fetches by ignoring obsolete notifications.

Modifications
-------------
Process topology update notifications one at a time:

* Serialize the topology update signal processing by using
`concatMap` instead of `flatMap`.

* Before fetching new topology, compare current topology against
the latest version the server told us about. If we have up-to-date
topology, ignore the obsolete notification.

* Call `provider.propose{Bucket,Global}Config` in `doOnNext`
instead of in an external subscriber. This ensures the SDK's
"current" topology revision is updated immediately, before
processing the next signal.

--

Add `TopologyChangeNotificationBuffer` for storing notifications.
Only ask server for latest topology if a notification is still relevant.

Use this separate buffer for the topology revisions because
storing them in a Flux would be excessively complex.

--

Move the topology polling trigger flux creation code into
DefaultConfigurationProvider, so it can be shared by the
Global and KV refreshers. (The flux itself isn't shared --
only the code to create it).

For consistency, and to ensure we don't block the KV threads,
handle topology change notifications in the same thread as
timer ticks (instead of in the KV thread).

Change-Id: Icabc7b10eabea6977a376b931f4e4c43f05350bb
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/235174
Tested-by: Build Bot <[email protected]>
Reviewed-by: Michael Reiche <[email protected]>

Author: dnault

core-io/src/integrationTest/java/com/couchbase/client/core/config/refresher/ProposedBucketConfigInspectingProvider.java

@@ -8,10 +8,12 @@
 import com.couchbase.client.core.env.SeedNode;
 import com.couchbase.client.core.io.CollectionIdentifier;
 import com.couchbase.client.core.io.CollectionMap;
+import com.couchbase.client.core.topology.TopologyRevision;
+import org.jspecify.annotations.Nullable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.util.Collections;
+import java.time.Duration;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -133,11 +135,16 @@ public void signalConfigRefreshFailed(ConfigRefreshFailure failure) {
   }
 
   @Override
-  public void signalConfigChanged() {
+  public void signalNewTopologyAvailable(@Nullable String bucketName, @Nullable TopologyRevision availableRevision) {
   }
 
   @Override
-  public Flux<Long> configChangeNotifications() {
-    return Flux.empty();
+  public @Nullable TopologyRevision removeTopologyRevisionChangeNotification(@Nullable String bucketName) {
+    return null;
+  }
+
+  @Override
+  public Flux<TopologyPollingTrigger> topologyPollingTriggers(Duration timerInterval) {
+    return delegate.topologyPollingTriggers(timerInterval);
   }
 }

core-io/src/main/java/com/couchbase/client/core/config/ClusterConfig.java

@@ -20,6 +20,7 @@
 import com.couchbase.client.core.service.ServiceType;
 import com.couchbase.client.core.topology.ClusterTopology;
 import com.couchbase.client.core.topology.ClusterTopologyWithBucket;
+import com.couchbase.client.core.topology.TopologyRevision;
 import reactor.util.annotation.Nullable;
 
 import java.util.Collection;
@@ -77,6 +78,16 @@ public Collection<ClusterTopologyWithBucket> bucketTopologies() {
     return g == null ? null : g.asClusterTopology();
   }
 
+  public TopologyRevision globalTopologyRevision() {
+    ClusterTopology t = globalTopology();
+    return t == null ? TopologyRevision.ZERO : t.revision();
+  }
+
+  public TopologyRevision bucketTopologyRevision(String bucketName) {
+    ClusterTopology t = bucketTopology(bucketName);
+    return t == null ? TopologyRevision.ZERO : t.revision();
+  }
+
   public BucketConfig bucketConfig(final String bucketName) {
     return bucketConfigs.get(bucketName);
   }

core-io/src/main/java/com/couchbase/client/core/config/ConfigurationProvider.java

@@ -20,9 +20,12 @@
 import com.couchbase.client.core.env.SeedNode;
 import com.couchbase.client.core.io.CollectionIdentifier;
 import com.couchbase.client.core.io.CollectionMap;
+import com.couchbase.client.core.topology.TopologyRevision;
+import org.jspecify.annotations.Nullable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.time.Duration;
 import java.util.Set;
 
 /**
@@ -170,17 +173,35 @@ default void republishCurrentConfig() {
   /**
    * Signals to the config provider that the server sent a notification
    * that the cluster topology or bucket config changed.
+   *
+   * @param bucketName null if the signal is for the global topology
+   * @param availableRevision null if the server did not tell us the new revision
    */
-  void signalConfigChanged();
+  void signalNewTopologyAvailable(
+    @Nullable String bucketName,
+    @Nullable TopologyRevision availableRevision
+  );
 
   /**
-   * Returns a feed that emits {@value TRIGGERED_BY_CONFIG_CHANGE_NOTIFICATION}
-   * whenever someone calls {@link #signalConfigChanged()}.
+   * Returns a feed of polling triggers, a combination of timer ticks and server notifications.
    */
-  Flux<Long> configChangeNotifications();
+  Flux<TopologyPollingTrigger> topologyPollingTriggers(Duration timerInterval);
+
+  enum TopologyPollingTrigger {
+    TIMER,
+    SERVER_NOTIFICATION,
+  }
 
   /**
-   * The value emitted by the {@link #configChangeNotifications()} feed.
+   * Returns and removes the latest unprocessed topology revision change notification
+   * for the specified context (bucket/global), or null if there are no
+   * unprocessed notifications for the specified context.
+   * <p>
+   * Might return the greatest possible topology revision if the server told us
+   * our topology was obsolete but didn't tell us the latest revision.
+   *
+   * @param bucketName null means global
    */
-  long TRIGGERED_BY_CONFIG_CHANGE_NOTIFICATION = -1L;
+  @Nullable
+  TopologyRevision removeTopologyRevisionChangeNotification(@Nullable String bucketName);
 }

core-io/src/main/java/com/couchbase/client/core/config/DefaultConfigurationProvider.java

@@ -68,6 +68,7 @@
 import com.couchbase.client.core.topology.NodeIdentifier;
 import com.couchbase.client.core.topology.PortSelector;
 import com.couchbase.client.core.topology.TopologyParser;
+import com.couchbase.client.core.topology.TopologyRevision;
 import com.couchbase.client.core.util.ConnectionString;
 import com.couchbase.client.core.util.ConnectionStringUtil;
 import com.couchbase.client.core.util.NanoTimestamp;
@@ -79,6 +80,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.util.annotation.Nullable;
 import reactor.util.retry.Retry;
@@ -100,6 +102,7 @@
 import static com.couchbase.client.core.Reactor.ignoreIfDone;
 import static com.couchbase.client.core.logging.RedactableArgument.redactMeta;
 import static com.couchbase.client.core.logging.RedactableArgument.redactSystem;
+import static com.couchbase.client.core.util.CbStrings.nullToEmpty;
 import static com.couchbase.client.core.util.ConnectionStringUtil.fromDnsSrvOrThrowIfTlsRequired;
 import static java.util.Collections.emptySet;
 import static java.util.Objects.requireNonNull;
@@ -185,7 +188,8 @@ public class DefaultConfigurationProvider implements ConfigurationProvider {
 
   private volatile NanoTimestamp lastDnsSrvLookup = NanoTimestamp.never();
 
-  private final Sinks.Many<Long> configPollTrigger = Sinks.many().multicast().directBestEffort();
+  private final Sinks.Many<TopologyPollingTrigger> topologyPollingTriggers = Sinks.many().multicast().onBackpressureBuffer(1);
+  private final TopologyChangeNotificationBuffer topologyChangeNotificationBuffer = new TopologyChangeNotificationBuffer();
 
   /**
    * Creates a new configuration provider.
@@ -812,14 +816,46 @@ public void signalConfigRefreshFailed(final ConfigRefreshFailure failure) {
     }
   }
 
+  private static final TopologyRevision NEWER_THAN_WHAT_WE_HAVE = new TopologyRevision(Long.MAX_VALUE, Long.MAX_VALUE) {
+    @Override
+    public String toString() {
+      return "?.?";
+    }
+  };
+
   @Override
-  public synchronized void signalConfigChanged() {
-    configPollTrigger.tryEmitNext(TRIGGERED_BY_CONFIG_CHANGE_NOTIFICATION);
+  public void signalNewTopologyAvailable(
+    @Nullable String bucketName,
+    @Nullable TopologyRevision availableRevision
+  ) {
+
+    if (availableRevision == null) {
+      availableRevision = NEWER_THAN_WHAT_WE_HAVE;
+    }
+
+    if (topologyChangeNotificationBuffer.putIfNewer(nullToEmpty(bucketName), availableRevision)) {
+      // synchronized to prevent tryEmitNext from returning EmitResult.FAIL_NON_SERIALIZED
+      synchronized (this) {
+        topologyPollingTriggers.tryEmitNext(TopologyPollingTrigger.SERVER_NOTIFICATION);
+      }
+    }
   }
 
   @Override
-  public Flux<Long> configChangeNotifications() {
-    return configPollTrigger.asFlux();
+  public Flux<TopologyPollingTrigger> topologyPollingTriggers(Duration timerInterval) {
+    Scheduler scheduler = core.context().environment().scheduler();
+
+    return Flux.merge(
+      Flux.interval(timerInterval, core.context().environment().scheduler())
+        .onBackpressureDrop() // otherwise the `interval` operator signals an error
+        .map(it -> TopologyPollingTrigger.TIMER),
+
+      topologyPollingTriggers.asFlux().publishOn(scheduler)
+    );
+  }
+
+  public @Nullable TopologyRevision removeTopologyRevisionChangeNotification(@Nullable String bucketName) {
+    return topologyChangeNotificationBuffer.remove(bucketName);
   }
 
   /**

core-io/src/main/java/com/couchbase/client/core/config/TopologyChangeNotificationBuffer.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2025 Couchbase, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.couchbase.client.core.config;
+
+import com.couchbase.client.core.annotation.Stability;
+import com.couchbase.client.core.topology.TopologyRevision;
+import org.jspecify.annotations.NullMarked;
+import org.jspecify.annotations.Nullable;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import static com.couchbase.client.core.util.CbStrings.nullToEmpty;
+
+/**
+ * "Squashes" notifications, retaining only the highest revision for each bucket (plus global).
+ */
+@NullMarked
+@Stability.Internal
+public class TopologyChangeNotificationBuffer {
+  private final ConcurrentMap<String, TopologyRevision> bucketNameToRevision = new ConcurrentHashMap<>();
+
+  /**
+   * Returns true if the given revision was added to the buffer,
+   * or false if it was not newer than the existing entry.
+   *
+   * @param bucketName null means global
+   */
+  public boolean putIfNewer(
+    @Nullable String bucketName,
+    TopologyRevision revision
+  ) {
+    TopologyRevision updatedRevision = bucketNameToRevision.compute(
+      key(bucketName),
+      (k, existing) ->
+        existing == null || revision.newerThan(existing)
+          ? revision
+          : existing
+    );
+
+    return revision == updatedRevision;
+  }
+
+  /**
+   * @param bucketName null means global
+   */
+  public @Nullable TopologyRevision remove(@Nullable String bucketName) {
+    return bucketNameToRevision.remove(key(bucketName));
+  }
+
+  private static String key(@Nullable String bucketName) {
+    return nullToEmpty(bucketName);
+  }
+
+  @Override
+  public String toString() {
+    return "TopologyChangeNotificationBuffer{" +
+      "bucketNameToRevision=" + bucketNameToRevision +
+      '}';
+  }
+}

core-io/src/main/java/com/couchbase/client/core/config/refresher/GlobalRefresher.java

@@ -23,13 +23,17 @@
 import com.couchbase.client.core.config.ConfigRefreshFailure;
 import com.couchbase.client.core.config.ConfigVersion;
 import com.couchbase.client.core.config.ConfigurationProvider;
+import com.couchbase.client.core.config.ConfigurationProvider.TopologyPollingTrigger;
 import com.couchbase.client.core.config.GlobalConfig;
 import com.couchbase.client.core.config.PortInfo;
 import com.couchbase.client.core.config.ProposedGlobalConfigContext;
 import com.couchbase.client.core.msg.kv.CarrierGlobalConfigRequest;
 import com.couchbase.client.core.retry.FailFastRetryStrategy;
 import com.couchbase.client.core.service.ServiceType;
+import com.couchbase.client.core.topology.TopologyRevision;
 import com.couchbase.client.core.util.NanoTimestamp;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -42,7 +46,6 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
-import static com.couchbase.client.core.config.DefaultConfigurationProvider.TRIGGERED_BY_CONFIG_CHANGE_NOTIFICATION;
 import static com.couchbase.client.core.config.refresher.KeyValueBucketRefresher.MAX_PARALLEL_FETCH;
 import static com.couchbase.client.core.config.refresher.KeyValueBucketRefresher.POLLER_INTERVAL;
 import static com.couchbase.client.core.config.refresher.KeyValueBucketRefresher.clampConfigRequestTimeout;
@@ -56,6 +59,7 @@
  * needed.
  */
 public class GlobalRefresher {
+  private final Logger log = LoggerFactory.getLogger(GlobalRefresher.class);
 
   /**
    * Holds the parent configuration provider.
@@ -116,28 +120,49 @@ public GlobalRefresher(final ConfigurationProvider provider, final Core core) {
     configPollInterval = core.context().environment().ioConfig().configPollInterval();
     configRequestTimeout = clampConfigRequestTimeout(configPollInterval);
 
-    pollRegistration = Flux.merge(
-        Flux.interval(pollerInterval(), core.context().environment().scheduler()),
-        provider.configChangeNotifications()
-      )
-      // Proposing a new config should be quick, but just in case it gets held up we do
-      // not want to terminate the refresher and just drop the ticks and keep going.
-      .onBackpressureDrop()
-      // If the refresher is not started yet, drop all intervals
-      .filter(v -> started)
-      // Since the POLLER_INTERVAL is smaller than the config poll interval, make sure
-      // we only emit poll events if enough time has elapsed -- or if the server told us
-      // there's a new config.
-      .filter(v -> v == TRIGGERED_BY_CONFIG_CHANGE_NOTIFICATION || lastPoll.hasElapsed(configPollInterval))
-      .flatMap(ign -> {
+    pollRegistration = provider.topologyPollingTriggers(pollerInterval())
+      .filter(trigger -> started)
+      .concatMap(trigger -> {
+        if (trigger == TopologyPollingTrigger.TIMER) {
+          // It's a polling tick! Ticks can be more frequent than the configured polling interval, or might occur right after a triggered update, so maybe skip this tick.
+          if (!lastPoll.hasElapsed(configPollInterval)) {
+            log.trace("Ignoring tick because last poll for global topology was {} ago, which is less than config poll interval {}", lastPoll.elapsed(), configPollInterval);
+            return Mono.empty();
+
+          } else {
+            log.debug("Polling for global topology because polling interval has elapsed.");
+          }
+
+        } else if (trigger == TopologyPollingTrigger.SERVER_NOTIFICATION) {
+          TopologyRevision availableRevision = provider.removeTopologyRevisionChangeNotification(null);
+          TopologyRevision currentRevision = provider.config().globalTopologyRevision();
+
+          if (availableRevision == null) {
+            log.trace("Ignoring redundant global topology change notification.");
+            return Mono.empty();
+          }
+
+          if (availableRevision.newerThan(currentRevision)) {
+            log.debug("Fetching updated global topology; available revision {} is newer than current ({}).", availableRevision, currentRevision);
+
+          } else {
+            log.debug("Skipping global topology revision {} because it's not newer than current ({}).", availableRevision, currentRevision);
+            return Mono.empty();
+          }
+
+        } else {
+          return Mono.error(new RuntimeException("Unexpected topology poll trigger: " + trigger));
+        }
+
         List<PortInfo> nodes = filterEligibleNodes();
         if (numFailedRefreshes.get() >= nodes.size()) {
           provider.signalConfigRefreshFailed(ConfigRefreshFailure.ALL_NODES_TRIED_ONCE_WITHOUT_SUCCESS);
           numFailedRefreshes.set(0);
         }
-        return attemptUpdateGlobalConfig(Flux.fromIterable(nodes).take(MAX_PARALLEL_FETCH));
+        return attemptUpdateGlobalConfig(Flux.fromIterable(nodes).take(MAX_PARALLEL_FETCH))
+          .doOnNext(provider::proposeGlobalConfig);
       })
-      .subscribe(provider::proposeGlobalConfig);
+      .subscribe();
   }
 
   /**