Fixed - destroy() method doesn't work if called immediately after creation of RDelayedQueue object. #5537

redisson/src/main/java/org/redisson/QueueTransferTask.java

@@ -58,7 +58,7 @@ public abstract class QueueTransferTask {
         
     }
     
-    private int usage = 1;
+    private volatile int usage = 1;
     private final AtomicReference<TimeoutTask> lastTimeout = new AtomicReference<TimeoutTask>();
     private final ServiceManager serviceManager;
 
@@ -99,14 +99,23 @@ public abstract class QueueTransferTask {
     public void stop() {
         RTopic schedulerTopic = getTopic();
         schedulerTopic.removeListener(messageListenerId, statusListenerId);
+
+        TimeoutTask oldTimeout = lastTimeout.get();
+        if (oldTimeout != null) {
+            oldTimeout.getTask().cancel();
+        }
     }
 
     private void scheduleTask(final Long startTime) {
-        TimeoutTask oldTimeout = lastTimeout.get();
+        if (usage == 0) {
+            return;
+        }
+
         if (startTime == null) {
             return;
         }
 
+        TimeoutTask oldTimeout = lastTimeout.get();
         if (oldTimeout != null) {
             oldTimeout.getTask().cancel();
         }
@@ -137,6 +146,10 @@ public abstract class QueueTransferTask {
     protected abstract RFuture<Long> pushTaskAsync();
     
     private void pushTask() {
+        if (usage == 0) {
+            return;
+        }
+
         RFuture<Long> startTimeFuture = pushTaskAsync();
         startTimeFuture.whenComplete((res, e) -> {
             if (e != null) {

redisson/src/test/java/org/redisson/RedissonDelayedQueueTest.java

@@ -1,6 +1,7 @@
 package org.redisson;
 
 import org.junit.jupiter.api.Test;
+import org.redisson.api.RBlockingDeque;
 import org.redisson.api.RBlockingQueue;
 import org.redisson.api.RDelayedQueue;
 import org.redisson.api.RQueue;
@@ -12,6 +13,22 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 public class RedissonDelayedQueueTest extends RedisDockerTest {
 
+    @Test
+    public void testDestroy() throws InterruptedException {
+        RBlockingDeque<Integer> blockingDeque = redisson.getBlockingDeque("test");
+        RDelayedQueue<Integer> delayedQueue = redisson.getDelayedQueue(blockingDeque);
+        delayedQueue.offer(1, 2, TimeUnit.SECONDS);
+        delayedQueue.destroy();
+
+        RDelayedQueue<Integer> delayedQueue2 = redisson.getDelayedQueue(blockingDeque);
+        delayedQueue2.destroy();
+
+        Thread.sleep(1000);
+
+        Object s = blockingDeque.poll(3, TimeUnit.SECONDS);
+        assertThat(s).isNull();
+    }
+
     @Test
     public void testRemove() throws InterruptedException {
         RBlockingQueue<String> blockingFairQueue = redisson.getBlockingQueue("delay_queue");