Merge pull request #1438 from smallrye/fix/1436

fix: do not interrupt threads on cancellation in UniDelayOnItem
smallrye · Nov 22, 2023 · b52c5a5 · b52c5a5
2 parents 0217652 + d883a75
commit b52c5a5
Show file tree

Hide file tree

Showing 8 changed files with 25 additions and 7 deletions.
diff --git a/...ementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiBufferWithTimeoutOp.java b/...ementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiBufferWithTimeoutOp.java
@@ -236,7 +236,7 @@ public void request(long n) {
         public void onCompletion() {
             if (terminated.compareAndSet(RUNNING, SUCCEED)) {
                 if (task != null) {
-                    task.cancel(true);
+                    task.cancel(false);
                     task = null;
                 }
                 checkedComplete();

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiDemandPacer.java b/implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiDemandPacer.java
@@ -136,7 +136,7 @@ public void onCompletion() {
         @Override
         public void cancel() {
             if (scheduledFuture != null) {
-                scheduledFuture.cancel(true);
+                scheduledFuture.cancel(false);
             }
             super.cancel();
         }

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiWindowOnDurationOp.java b/implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiWindowOnDurationOp.java
@@ -267,7 +267,7 @@ boolean replace(Future<?> task) {
                 Future current = container.get();
                 if (current == NONE) {
                     if (task != null) {
-                        task.cancel(true);
+                        task.cancel(false);
                     }
                     return false;
                 }

diff --git a/...ation/src/main/java/io/smallrye/mutiny/operators/multi/multicast/MultiReferenceCount.java b/...ation/src/main/java/io/smallrye/mutiny/operators/multi/multicast/MultiReferenceCount.java
@@ -75,7 +75,7 @@ void cancel(ConnectableMultiConnection connection) {
         }
 
         ScheduledFuture<?> future = executor.schedule(connection, duration.toMillis(), TimeUnit.MILLISECONDS);
-        connection.setTimer(() -> future.cancel(true));
+        connection.setTimer(() -> future.cancel(false));
     }
 
     void terminated(ConnectableMultiConnection connection) {

diff --git a/implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniDelayOnItem.java b/implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniDelayOnItem.java
@@ -41,7 +41,7 @@ public void cancel() {
             if (!isCancelled()) {
                 super.cancel();
                 if (scheduledFuture != null) {
-                    scheduledFuture.cancel(true);
+                    scheduledFuture.cancel(false);
                 }
             }
         }

diff --git a/implementation/src/test/java/io/smallrye/mutiny/groups/UniOnFailureRetryTest.java b/implementation/src/test/java/io/smallrye/mutiny/groups/UniOnFailureRetryTest.java
@@ -9,6 +9,7 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -558,4 +559,21 @@ public void rejectNullExecutors() {
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("`executor` must not be `null`");
     }
+
+    @Test
+    void avoidSpuriousInterruptsWithBackoff() {
+        AtomicBoolean flip = new AtomicBoolean();
+        AtomicBoolean interrupted = new AtomicBoolean();
+        String result = Uni.createFrom().item(() -> {
+            if (flip.getAndSet(true)) {
+                return "yolo";
+            }
+            throw new RuntimeException("boom");
+        })
+                .onFailure().retry().withBackOff(Duration.ofMillis(100), Duration.ofSeconds(5)).atMost(1)
+                .onItem().invoke(() -> interrupted.set(Thread.currentThread().isInterrupted()))
+                .await().atMost(Duration.ofSeconds(1));
+        assertThat(result).isEqualTo("yolo");
+        assertThat(interrupted).isFalse();
+    }
 }
diff --git a/implementation/src/test/java/io/smallrye/mutiny/operators/MultiIgnoreTest.java b/implementation/src/test/java/io/smallrye/mutiny/operators/MultiIgnoreTest.java
@@ -69,7 +69,7 @@ public void testAsUniWithNever() {
                 .onItem().ignoreAsUni().subscribeAsCompletionStage();
 
         assertThat(future).isNotCompleted();
-        future.cancel(true);
+        future.cancel(false);
     }
 
     @Test

diff --git a/implementation/src/test/java/io/smallrye/mutiny/operators/UniCreateFromFutureTest.java b/implementation/src/test/java/io/smallrye/mutiny/operators/UniCreateFromFutureTest.java
@@ -371,7 +371,7 @@ public void testWithInterruptedFuture() {
         UniAssertSubscriber<String> subscriber = UniAssertSubscriber.create();
         CompletableFuture<String> cs = new CompletableFuture<>();
         Uni.createFrom().future(cs).subscribe().withSubscriber(subscriber);
-        cs.cancel(true);
+        cs.cancel(false);
         subscriber
                 .awaitFailure()
                 .assertFailedWith(CancellationException.class, null);