github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Fixed - RScheduledExecutorService cron triggers fire continuously for hours for some time zones (regression since 3.16.5)

Browse Source
pull/4608/head
Nikita Koksharov 2 years ago
parent b6e9ec2a85
commit 2e874b085a
5 changed files with 1379 additions and 601 deletions
Show Stats Download Patch File Download Diff File

11
redisson/src/main/java/org/redisson/RedissonExecutorService.java
Unescape Escape View File

@ -40,8 +40,6 @@ import java.lang.ref.ReferenceQueue;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZonedDateTime;
import java.util.*;
import java.util.concurrent.*;
@ -1218,12 +1216,11 @@ public class RedissonExecutorService implements RScheduledExecutorService {
check(task);
ClassBody classBody = getClassBody(task);
byte[] state = encode(task);
ZonedDateTime currentDate = ZonedDateTime.of(LocalDateTime.now(), cronSchedule.getZoneId());
ZonedDateTime startDate = cronSchedule.getExpression().nextTimeAfter(currentDate);
Date startDate = cronSchedule.getExpression().getNextValidTimeAfter(new Date());
if (startDate == null) {
throw new IllegalArgumentException("Wrong cron expression! Unable to calculate start date");
}
long startTime = startDate.toInstant().toEpochMilli();
long startTime = startDate.getTime();
String taskId = id;
ScheduledCronExpressionParameters params = new ScheduledCronExpressionParameters(taskId);
@ -1232,14 +1229,14 @@ public class RedissonExecutorService implements RScheduledExecutorService {
params.setLambdaBody(classBody.getLambda());
params.setState(state);
params.setStartTime(startTime);
params.setCronExpression(cronSchedule.getExpression().getExpr());
params.setCronExpression(cronSchedule.getExpression().getCronExpression());
params.setTimezone(cronSchedule.getZoneId().toString());
params.setExecutorId(executorId);
RemotePromise<Void> result = (RemotePromise<Void>) asyncScheduledServiceAtFixed.schedule(params).toCompletableFuture();
addListener(result);
RedissonScheduledFuture<Void> f = new RedissonScheduledFuture<Void>(result, startTime) {
public long getDelay(TimeUnit unit) {
return unit.convert(startDate.toInstant().toEpochMilli() - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
return unit.convert(startDate.getTime() - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
};
};
storeReference(f, result.getRequestId());

8
redisson/src/main/java/org/redisson/api/CronSchedule.java
Unescape Escape View File

@ -16,9 +16,9 @@
package org.redisson.api;
import org.redisson.executor.CronExpression;
import org.redisson.executor.CronExpressionEx;
import java.time.ZoneId;
import java.util.TimeZone;
/**
* Cron expression object used in {@link RScheduledExecutorService}.
@ -49,7 +49,7 @@ public final class CronSchedule {
* wrapping a ParseException if the expression is invalid
*/
public static CronSchedule of(String expression) {
return new CronSchedule(new CronExpressionEx(expression), ZoneId.systemDefault());
return of(expression, ZoneId.systemDefault());
}
/**
@ -62,7 +62,9 @@ public final class CronSchedule {
* wrapping a ParseException if the expression is invalid
*/
public static CronSchedule of(String expression, ZoneId zoneId) {
return new CronSchedule(new CronExpressionEx(expression), zoneId);
CronExpression ce = new CronExpression(expression);
ce.setTimeZone(TimeZone.getTimeZone(zoneId));
return new CronSchedule(ce, zoneId);
}
/**

1877
redisson/src/main/java/org/redisson/executor/CronExpression.java
View File

File diff suppressed because it is too large Load Diff

71
redisson/src/main/java/org/redisson/executor/CronExpressionEx.java
Unescape Escape View File

@ -1,71 +0,0 @@
/**
* Copyright (c) 2013-2022 Nikita Koksharov
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.redisson.executor;
import java.time.ZonedDateTime;
import java.util.Arrays;
import java.util.NavigableSet;
import java.util.TreeSet;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class CronExpressionEx extends CronExpression {
private static final ThreadLocal<NavigableSet<Integer>> YEARS_FIELD = new ThreadLocal<>();
private final NavigableSet<Integer> years;
public CronExpressionEx(String expr) {
super(parseYear(expr));
years = YEARS_FIELD.get();
YEARS_FIELD.remove();
}
public CronExpressionEx(String expr, boolean withSeconds) {
super(parseYear(expr), withSeconds);
years = YEARS_FIELD.get();
YEARS_FIELD.remove();
}
@Override
public ZonedDateTime nextTimeAfter(ZonedDateTime afterTime) {
if (years != null) {
afterTime = afterTime.withYear(years.ceiling(afterTime.getYear()));
}
return super.nextTimeAfter(afterTime);
}
private static String parseYear(String expr) {
String[] parts = expr.split("\\s+");
if (parts.length == 7) {
String year = parts[6];
String[] years = year.split(",");
if (years.length > 1) {
NavigableSet<Integer> yy = new TreeSet<>(Arrays.stream(years).map(y -> Integer.valueOf(y))
.collect(Collectors.toList()));
YEARS_FIELD.set(yy);
}
String[] yearsRange = year.split("-");
if (yearsRange.length > 1) {
NavigableSet<Integer> yy = new TreeSet<>(IntStream.rangeClosed(Integer.valueOf(yearsRange[0]), Integer.valueOf(yearsRange[1]))
.boxed().collect(Collectors.toList()));
YEARS_FIELD.set(yy);
}
return expr.replace(year, "").trim();
}
return expr;
}
}

13
redisson/src/main/java/org/redisson/executor/TasksRunnerService.java
Unescape Escape View File

@ -38,11 +38,10 @@ import org.redisson.remote.ResponseEntry;
import java.io.ByteArrayInputStream;
import java.io.ObjectInput;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.Arrays;
import java.util.Date;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
@ -137,13 +136,13 @@ public class TasksRunnerService implements RemoteExecutorService {
@Override
public void schedule(ScheduledCronExpressionParameters params) {
CronExpression expression = new CronExpressionEx(params.getCronExpression());
ZonedDateTime currentDate = ZonedDateTime.of(LocalDateTime.now(), ZoneId.of(params.getTimezone()));
ZonedDateTime nextStartDate = expression.nextTimeAfter(currentDate);
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.toInstant().toEpochMilli());
params.setStartTime(nextStartDate.getTime());
future = service.schedule(params);
}
try {

Write Preview
Loading…
Cancel
Save