Bug fix: #5148

Signed-off-by: Guozhang Wu <wguozhang@vmware.com>

redisson/src/main/java/org/redisson/executor/TasksRunnerService.java

@@ -139,21 +139,15 @@ public class TasksRunnerService implements RemoteExecutorService {
         CronExpression expression = new CronExpression(params.getCronExpression());
         expression.setTimeZone(TimeZone.getTimeZone(params.getTimezone()));
         Date nextStartDate = expression.getNextValidTimeAfter(new Date());
-        RFuture<Void> future = null;
-        if (nextStartDate != null) {
-            RemoteExecutorServiceAsync service = asyncScheduledServiceAtFixed(params.getExecutorId(), params.getRequestId());
-            params.setStartTime(nextStartDate.getTime());
-            future = service.schedule(params);
-        }
-        try {
-            executeRunnable(params, nextStartDate == null);
-        } catch (Exception e) {
-            // cancel task if it throws an exception
-            if (future != null) {
-                future.cancel(true);
-            }
-            throw e;
+
+        executeRunnable(params, nextStartDate == null);
+
+        if (nextStartDate == null || !redisson.getMap(tasksName, StringCodec.INSTANCE).containsKey(params.getRequestId())) {
+            return;
         }
+
+        params.setStartTime(nextStartDate.getTime());
+        asyncScheduledServiceAtFixed(params.getExecutorId(), params.getRequestId()).schedule(params);
     }
 
     /**

redisson/src/test/java/org/redisson/executor/RedissonScheduledExecutorServiceTest.java

@@ -392,6 +392,41 @@ public class RedissonScheduledExecutorServiceTest extends BaseTest {
         assertThat(redisson.getAtomicLong("executed2").get()).isEqualTo(30);
     }
 
+    @Test
+    public void testCancelCronExpression() throws InterruptedException, ExecutionException {
+        RScheduledExecutorService executor = redisson.getExecutorService("test");
+        RScheduledFuture<?> future = executor.schedule(new ScheduledRunnableTask("executed"), CronSchedule.of("0/2 * * * * ?"));
+        Thread.sleep(TimeUnit.SECONDS.toMillis(10));
+        assertThat(redisson.getAtomicLong("executed").get()).isEqualTo(5);
+
+        cancel(future);
+
+        Thread.sleep(TimeUnit.SECONDS.toMillis(3));
+        assertThat(redisson.getAtomicLong("executed").get()).isEqualTo(5);
+
+        executor.delete();
+        redisson.getKeys().delete("executed");
+        assertThat(redisson.getKeys().count()).isZero();
+    }
+
+    @Test
+    public void testCancelAndInterruptCronExpression() throws InterruptedException, ExecutionException {
+        RScheduledExecutorService executor = redisson.getExecutorService("test");
+        RScheduledFuture<?> future = executor.schedule(new ScheduledLongRepeatableTask("counter", "executed"), CronSchedule.of("0/2 * * * * ?"));
+        Thread.sleep(TimeUnit.SECONDS.toMillis(6));
+        assertThat(redisson.getAtomicLong("counter").get()).isEqualTo(3);
+
+        cancel(future);
+        assertThat(redisson.<Long>getBucket("executed").get()).isBetween(1000L, Long.MAX_VALUE);
+
+        Thread.sleep(TimeUnit.SECONDS.toMillis(3));
+        assertThat(redisson.getAtomicLong("counter").get()).isEqualTo(3);
+
+        executor.delete();
+        redisson.getKeys().delete("counter", "executed");
+        assertThat(redisson.getKeys().count()).isZero();
+    }
+
     public static class RunnableTask2 implements Runnable, Serializable {
 
         @Override