JVMCBC-1705 Some uses of Flux.interval() are not protected by onBackpressureDrop

Motivation
----------
Users in serverless execution environments have seen
SDK Flux.interval() operators failing due to lack of
downstream requests. This could potentially terminate
transaction cleanup.

Modifications
-------------
Replace most calls to Flux.interval() with new method
Reactor.safeInterval() that always drops on backpressure.

Add Reactor.unsafeInterval() as an alias for Flux.interval(),
and use it in LostCleanupDistributed where overflow from
Flux.interval() is expected for correct behavior.

Upgrade to latest Checkstyle. We can do this because
a while back we started requiring Java 17 at build time.

Add Checkstyle rule to prohibit dangerous
usage of Flux.interval().

Change-Id: I66a799a6957d2b010bf5d2205bf8de082b0cf9f8
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/236256
Tested-by: Build Bot <[email protected]>
Reviewed-by: Graham Pople <[email protected]>

Author: dnault

config/checkstyle/checkstyle-basic.xml

@@ -11,6 +11,18 @@
     <property name="fileExtensions" value="java,scala"/>
   </module>
 
+  <module name="RegexpMultiline">
+    <!-- Match static imports and qualified references. -->
+    <property name="format" value="Flux\s*\.\s*interval"/>
+    <property name="message" value="Please use Reactor.safeInterval() instead of Flux.interval()"/>
+  </module>
+
+  <module name="SuppressWithPlainTextCommentFilter">
+    <property name="offCommentFormat" value="CHECKSTYLE:OFF ([\w\|]+)"/>
+    <property name="onCommentFormat" value="CHECKSTYLE:ON ([\w\|]+)"/>
+    <property name="checkFormat" value="$1"/>
+  </module>
+
   <module name="TreeWalker">
     <module name="IllegalImport">
       <!-- Prevent unintentional dependency on unbundled Jackson -->

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

@@ -30,6 +30,8 @@
 import reactor.core.publisher.Operators;
 import reactor.core.publisher.SignalType;
 import reactor.core.publisher.Sinks;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 import reactor.util.context.Context;
 
 import java.time.Duration;
@@ -291,4 +293,39 @@ public static Sinks.EmitResult ignoreIfDone(Sinks.EmitResult result) {
       ? OK
       : result;
   }
+
+  /**
+   * Alias for {@link Flux#interval(Duration, Scheduler)} that doesn't trigger a Checkstyle error.
+   * <p>
+   * This operation is "unsafe" because if there is backpressure it emits the dreaded OverflowException: "interval doesn't support small downstream requests that replenish slower than the ticks".
+   */
+  @Stability.Internal
+  public static Flux<Long> unsafeInterval(Duration d, Scheduler scheduler) {
+    // CHECKSTYLE:OFF RegexpMultiline - Allow Flux.interval
+    return Flux.interval(d, scheduler);
+    // CHECKSTYLE:ON RegexpMultiline
+  }
+
+  /**
+   * Like {@link Flux#interval(Duration, Scheduler)}, but drops signals (instead of exploding) when there is insufficient demand.
+   * <p>
+   * Avoids the dreaded OverflowException: "interval doesn't support small downstream requests that replenish slower than the ticks".
+   */
+  @Stability.Internal
+  public static Flux<Long> safeInterval(Duration period, Scheduler scheduler) {
+    return safeInterval(period, period, scheduler);
+  }
+
+  /**
+   * Like {@link Flux#interval(Duration, Duration, Scheduler)}, but drops signals (instead of exploding) when there is insufficient demand.
+   * <p>
+   * Avoids the dreaded OverflowException: "interval doesn't support small downstream requests that replenish slower than the ticks".
+   */
+  @Stability.Internal
+  public static Flux<Long> safeInterval(Duration delay, Duration period, Scheduler scheduler) {
+    // CHECKSTYLE:OFF RegexpMultiline - Allow Flux.interval
+    return Flux.interval(delay, period, scheduler)
+      .onBackpressureDrop();
+    // CHECKSTYLE:ON RegexpMultiline
+  }
 }

core-io/src/main/java/com/couchbase/client/core/cnc/DefaultEventBus.java

@@ -20,7 +20,6 @@
 import com.couchbase.client.core.util.CbCollections;
 import com.couchbase.client.core.util.NanoTimestamp;
 import com.couchbase.client.core.util.NativeImageHelper;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
 
@@ -37,6 +36,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 
+import static com.couchbase.client.core.Reactor.safeInterval;
 import static com.couchbase.client.core.util.CbThrowables.getStackTraceAsString;
 
 /**
@@ -281,7 +281,7 @@ public Mono<Void> stop(final Duration timeout) {
         overflowInfo.clear();
         return Mono.empty();
       })
-      .then(Flux.interval(Duration.ofMillis(10), scheduler).takeUntil(i -> !runningThread.isAlive()).then())
+      .then(safeInterval(Duration.ofMillis(10), scheduler).takeUntil(i -> !runningThread.isAlive()).then())
       .timeout(timeout, scheduler);
   }
 

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

@@ -100,6 +100,7 @@
 
 import static com.couchbase.client.core.Reactor.emitFailureHandler;
 import static com.couchbase.client.core.Reactor.ignoreIfDone;
+import static com.couchbase.client.core.Reactor.safeInterval;
 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;
@@ -846,8 +847,7 @@ public Flux<TopologyPollingTrigger> topologyPollingTriggers(Duration timerInterv
     Scheduler scheduler = core.context().environment().scheduler();
 
     return Flux.merge(
-      Flux.interval(timerInterval, core.context().environment().scheduler())
-        .onBackpressureDrop() // otherwise the `interval` operator signals an error
+      safeInterval(timerInterval, core.context().environment().scheduler())
         .map(it -> TopologyPollingTrigger.TIMER),
 
       topologyPollingTriggers.asFlux().publishOn(scheduler)

core-io/src/main/java/com/couchbase/client/core/transaction/cleanup/CoreTransactionsCleanup.java

@@ -40,6 +40,7 @@
 import java.util.concurrent.DelayQueue;
 import java.util.concurrent.TimeUnit;
 
+import static com.couchbase.client.core.Reactor.safeInterval;
 import static com.couchbase.client.core.cnc.events.transaction.TransactionEvent.DEFAULT_CATEGORY;
 
 /**
@@ -127,7 +128,7 @@ private void runRegularAttemptsCleanupThread() {
         LOGGER_REGULAR.debug("Starting background cleanup thread to find transactions from this client");
 
         // Periodically check and drain the cleanupQueue
-        Flux.interval(Duration.ofMillis(100), core.context().environment().transactionsSchedulers().schedulerCleanup())
+        safeInterval(Duration.ofMillis(100), core.context().environment().transactionsSchedulers().schedulerCleanup())
 
                 .flatMap(v -> {
                     if (stop) {

core-io/src/main/java/com/couchbase/client/core/transaction/cleanup/LostCleanupDistributed.java

@@ -64,6 +64,8 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import static com.couchbase.client.core.Reactor.safeInterval;
+import static com.couchbase.client.core.Reactor.unsafeInterval;
 import static com.couchbase.client.core.io.CollectionIdentifier.DEFAULT_COLLECTION;
 import static com.couchbase.client.core.io.CollectionIdentifier.DEFAULT_SCOPE;
 import static com.couchbase.client.core.logging.RedactableArgument.redactMeta;
@@ -358,7 +360,7 @@ Mono<Void> createThreadForCollectionIfNeeded(CollectionIdentifier coll) {
     }
 
     private void periodicallyCheckCleanupSet() {
-        cleanupThreadLauncher = Flux.interval(Duration.ZERO, Duration.ofSeconds(1), core.context().environment().transactionsSchedulers().schedulerCleanup())
+        cleanupThreadLauncher = safeInterval(Duration.ZERO, Duration.ofSeconds(1), core.context().environment().transactionsSchedulers().schedulerCleanup())
                 .concatMap(v -> Flux.fromIterable(cleanupSet))
                 .publishOn(core.context().environment().transactionsSchedulers().schedulerCleanup())
                 .concatMap(this::createThreadForCollectionIfNeeded)
@@ -439,7 +441,7 @@ private Mono<Void> perCollectionThread(CollectionIdentifier collection) {
                     // handling the first ATR takes 0.3 seconds, the next should execute 0.7 seconds later.
                     // Similar if the first ATR takes > 1 second, the next should execute instantly.
                     return Flux.zip(Flux.fromIterable(atrsHandledByThisClient),
-                                    Flux.interval(Duration.ofNanos(checkAtrEveryNNanos)))
+                                    unsafeInterval(Duration.ofNanos(checkAtrEveryNNanos), core.context().environment().scheduler()))
                             .publishOn(core.context().environment().transactionsSchedulers().schedulerCleanup())
 
                         // Where the ATR cleanup magic happens

core-io/src/test/java/com/couchbase/client/core/config/refresher/GlobalRefresherTest.java

@@ -34,11 +34,12 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
 
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static com.couchbase.client.core.Reactor.safeInterval;
 import static com.couchbase.client.core.util.CbCollections.mapOf;
 import static com.couchbase.client.core.util.MockUtil.mockCore;
 import static com.couchbase.client.test.Util.waitUntilCondition;
@@ -78,7 +79,7 @@ static ConfigurationProvider mockConfigurationProvider(ClusterConfig clusterConf
     ConfigurationProvider provider = mock(ConfigurationProvider.class);
     when(provider.config()).thenReturn(clusterConfig);
     when(provider.topologyPollingTriggers(any())).
-      thenReturn(Flux.interval(FAST_CONFIG_POLL_INTERVAL).map(it -> TopologyPollingTrigger.TIMER));
+      thenReturn(safeInterval(FAST_CONFIG_POLL_INTERVAL, Schedulers.parallel()).map(it -> TopologyPollingTrigger.TIMER));
     return provider;
   }
 

core-io/src/test/java/com/couchbase/client/core/config/refresher/KeyValueBucketRefresherTest.java

@@ -34,11 +34,12 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
 
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static com.couchbase.client.core.Reactor.safeInterval;
 import static com.couchbase.client.core.util.CbCollections.mapOf;
 import static com.couchbase.client.core.util.MockUtil.mockCore;
 import static org.mockito.ArgumentMatchers.any;
@@ -83,7 +84,7 @@ void triggersEventIfAllNodesFailedToRefresh() {
     ClusterConfig clusterConfig = new ClusterConfig();
     when(provider.config()).thenReturn(clusterConfig);
     when(provider.topologyPollingTriggers(any()))
-      .thenReturn(Flux.interval(Duration.ofMillis(10)).map(it -> TopologyPollingTrigger.TIMER));
+      .thenReturn(safeInterval(Duration.ofMillis(10), Schedulers.parallel()).map(it -> TopologyPollingTrigger.TIMER));
 
     ClusterTopologyWithBucket config = new ClusterTopologyBuilder()
       .addNode("foo", node -> node.ports(mapOf(ServiceType.KV, 11210, ServiceType.MANAGER, 8091)))

pom.xml

@@ -349,13 +349,12 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-checkstyle-plugin</artifactId>
-                <version>3.1.2</version>
+                <version>3.6.0</version>
                 <dependencies>
                     <dependency>
                         <groupId>com.puppycrawl.tools</groupId>
                         <artifactId>checkstyle</artifactId>
-                        <!-- Last version to support Java 8 is 9.x -->
-                        <version>9.3</version>
+                        <version>12.1.1</version>
                     </dependency>
                 </dependencies>
                 <configuration>