From 1694dd04ae4060ec7e3511eef4b765ff1b1163d7 Mon Sep 17 00:00:00 2001
From: Nikita Koksharov <nkoksharov@redisson.pro>
Date: Fri, 8 Nov 2019 14:59:45 +0300
Subject: [PATCH] Fixed - Interrupted blocking methods aren't canceled #2403

---
 .../java/org/redisson/command/CommandAsyncService.java   | 9 +++++++--
 .../main/java/org/redisson/command/RedisExecutor.java    | 4 +++-
 .../java/org/redisson/RedissonBlockingQueueTest.java     | 9 +++++++--
 3 files changed, 17 insertions(+), 5 deletions(-)

diff --git a/redisson/src/main/java/org/redisson/command/CommandAsyncService.java b/redisson/src/main/java/org/redisson/command/CommandAsyncService.java
index f5fa274a1..e8b20d5ce 100644
--- a/redisson/src/main/java/org/redisson/command/CommandAsyncService.java
+++ b/redisson/src/main/java/org/redisson/command/CommandAsyncService.java
@@ -142,8 +142,13 @@ public class CommandAsyncService implements CommandAsyncExecutor {
     
     @Override
     public <V> V getInterrupted(RFuture<V> future) throws InterruptedException {
-        future.await();
-        
+        try {
+            future.await();
+        } catch (InterruptedException e) {
+            ((RPromise)future).tryFailure(e);
+            throw e;
+        }
+
         if (future.isSuccess()) {
             return future.getNow();
         }
diff --git a/redisson/src/main/java/org/redisson/command/RedisExecutor.java b/redisson/src/main/java/org/redisson/command/RedisExecutor.java
index 10f034cc9..eaa78f43b 100644
--- a/redisson/src/main/java/org/redisson/command/RedisExecutor.java
+++ b/redisson/src/main/java/org/redisson/command/RedisExecutor.java
@@ -372,7 +372,9 @@ public class RedisExecutor<V, R> {
             }
 
             // handling cancel operation for blocking commands
-            if (mainPromise.isCancelled() && !attemptPromise.isDone()) {
+            if ((mainPromise.isCancelled()
+                    || mainPromise.cause() instanceof  InterruptedException)
+                        && !attemptPromise.isDone()) {
                 log.debug("Canceled blocking operation {} used {}", command, connection);
                 connection.forceFastReconnectAsync().onComplete((r, ex) -> {
                     attemptPromise.cancel(true);
diff --git a/redisson/src/test/java/org/redisson/RedissonBlockingQueueTest.java b/redisson/src/test/java/org/redisson/RedissonBlockingQueueTest.java
index 572a97bdd..4e19d1fe7 100644
--- a/redisson/src/test/java/org/redisson/RedissonBlockingQueueTest.java
+++ b/redisson/src/test/java/org/redisson/RedissonBlockingQueueTest.java
@@ -295,12 +295,17 @@ public class RedissonBlockingQueueTest extends RedissonQueueTest {
                 }
             };
         };
-        
+
         t.start();
         t.join(1000);
-        
+
         t.interrupt();
         Awaitility.await().atMost(Duration.ONE_SECOND).untilTrue(interrupted);
+
+        RBlockingQueue<Integer> q = getQueue(redisson);
+        q.add(1);
+        Thread.sleep(1000);
+        assertThat(q.contains(1)).isTrue();
     }
 
     @Test