diff --git a/README.md b/README.md index 9c64c30e5..139b32cee 100644 --- a/README.md +++ b/README.md @@ -80,14 +80,14 @@ from [Maven central](https://search.maven.org/search?q=g:com.github.sonus21%20AN * Add dependency * Gradle ```groovy - implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.13.0-RELEASE' + implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.13.1-RELEASE' ``` * Maven ```xml com.github.sonus21 rqueue-spring-boot-starter - 2.13.0-RELEASE + 2.13.1-RELEASE ``` @@ -108,14 +108,14 @@ from [Maven central](https://search.maven.org/search?q=g:com.github.sonus21%20AN * Add Dependency * Gradle ```groovy - implementation 'com.github.sonus21:rqueue-spring:2.13.0-RELEASE' + implementation 'com.github.sonus21:rqueue-spring:2.13.1-RELEASE' ``` * Maven ```xml com.github.sonus21 rqueue-spring - 2.13.0-RELEASE + 2.13.1-RELEASE ``` * Add annotation `EnableRqueue` on application config class @@ -127,7 +127,6 @@ from [Maven central](https://search.maven.org/search?q=g:com.github.sonus21%20AN @EnableRqueue public class Application { - @Bean public RedisConnectionFactory redisConnectionFactory() { // return a redis connection factory @@ -321,13 +320,6 @@ to familiarise yourself with the project. **Please format your code with Google Java formatter.** -```groovy -// springBootVersion = '2.0.6.RELEASE' -// springVersion = '5.0.10.RELEASE' -// springDataVersion = '2.0.6.RELEASE' -// microMeterVersion = '1.1.0' -``` - ## Links * Documentation: [https://sonus21.github.io/rqueue](https://sonus21.github.io/rqueue) diff --git a/build.gradle b/build.gradle index 3ffcc0464..cd5f38c6f 100644 --- a/build.gradle +++ b/build.gradle @@ -25,11 +25,6 @@ allprojects { } ext { - springBootVersion = System.getenv("SPRING_BOOT_VERSION") - springVersion = System.getenv("SPRING_VERSION") - springDataVersion = System.getenv("SPRING_DATA_VERSION") - microMeterVersion = System.getenv("MICROMETER_VERSION") - springBootVersion = "3.0.1" springVersion = "6.0.3" springDataVersion = "3.0.0" @@ -77,7 +72,7 @@ ext { subprojects { group = "com.github.sonus21" - version = "3.0.1-RELEASE" + version = "3.1.0-RELEASE" dependencies { // https://mvnrepository.com/artifact/org.springframework/spring-messaging diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 77cf61875..fefffffda 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -8,22 +8,37 @@ layout: default All notable user-facing changes to this project are documented in this file. -## Release [3.0.1] 17-Jan-2022 +## Release [3.1.0] 24-June-2023 +{: .highlight } +Migrate to this version to reduce redis resource utilization + +This will fix an important bug happening due to task multiplications. This is causing more Redis +resource usage Please check #[193] + +## Release [2.13.1] 24-June-2023 +{: .highlight } +Migrate to this version to reduce redis resource utilization + +This will fix an important bug happening due to task multiplications. This is causing more Redis +resource usage Please check #[193] -We're so excited to release Rqueue `3.0.1`. This release supports the Java 17, Spring Boot 3.x and Spring Framework 6.x +## Release [3.0.1] 17-Jan-2022 + +We're so excited to release Rqueue `3.0.1`. This release supports the Java 17, Spring Boot 3.x and +Spring Framework 6.x ### [2.13.0] - 25-Dec-2022 ### Fixes -{: .highlight} + +{: .highlight} Migrate to this version as soon as possible to avoid duplicate message consumption post deletion. * Important fix for parallel message deletion or delete the message from message listener * No threads are available, improvement on message poller * Use System Zone ID for UI bottom screen - ### [2.12.0] - 14-Dec-2022 ### Fixes @@ -33,15 +48,14 @@ Migrate to this version as soon as possible to avoid duplicate message consumpti ### [2.11.1] - 18-Nov-2022 -{: .highlight} -Migrate to this version as soon as possible to avoid message build up. Messages in scheduled queue -can grow if poller is failing. Workaround is to restart the application. +{: .highlight} +Migrate to this version as soon as possible to avoid message build up. Messages in scheduled queue +can grow if poller is failing. Workaround is to restart the application. -* Message mover unreliability, scheduled message were not getting consumed once redis connection error occurs +* Message mover unreliability, scheduled message were not getting consumed once redis connection + error occurs * Upgraded Jquery version - - ### [2.10.2] - 16-Jul-2022 ### Fixes @@ -60,10 +74,10 @@ can grow if poller is failing. Workaround is to restart the application. ### [2.10.0] - 10-Oct-2021 -{: .warning } +{: .warning } Breaking change, if you're controlling any internal settings of Rqueue using application environment -or configuration variable than application can break. We've renamed some config keys, [see](./migration#290-to-210) - +or configuration variable than application can break. We've renamed some config +keys, [see](./migration#290-to-210) ### Fixes @@ -235,7 +249,6 @@ Breaking change, for migration [see](./migration#1x-to-2x) REDIS with prefix `rqueue-` then it will consider version 2. - Renamed annotation field `maxJobExecutionTime` to `visibilityTimeout` - ### Added - Web interface to visualize queue @@ -294,7 +307,6 @@ Breaking change, for migration [see](./migration#1x-to-2x) * The basic version of Asynchronous task execution using Redis for Spring and Spring Boot - [1.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/1.0-RELEASE [1.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/1.1-RELEASE @@ -353,6 +365,11 @@ Breaking change, for migration [see](./migration#1x-to-2x) [2.13.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.13.0-RELEASE +[2.13.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.13.1-RELEASE + [3.0.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/3.0.0-RELEASE +[3.1.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/3.1.0-RELEASE + [122]: https://github.com/sonus21/rqueue/issues/122 +[193]: https://github.com/sonus21/rqueue/issues/193 diff --git a/docs/index.md b/docs/index.md index 4f25d8c65..413fedd88 100644 --- a/docs/index.md +++ b/docs/index.md @@ -90,38 +90,38 @@ the latest version in your dependency manager. #### Spring Boot 2.x Setup -Add Rqueue Spring Boot Starter 2.9.0 and refresh your project. Once you've added the dependency, +Add Rqueue Spring Boot Starter 2.13.1 and refresh your project. Once you've added the dependency, you can start sending and consuming messages. * Gradle ```groovy - implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.9.0-RELEASE' + implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.13.1-RELEASE' ``` * Maven ```xml com.github.sonus21 rqueue-spring-boot-starter - 2.9.0-RELEASE + 2.13.1-RELEASE ``` #### Spring Boot 3.x Setup -Add Rqueue Spring Boot Starter 3.0.1 and refresh your project. Once you've added the dependency, +Add Rqueue Spring Boot Starter 3.1.0 and refresh your project. Once you've added the dependency, you can start sending and consuming messages. * Add dependency * Gradle ```groovy - implementation 'com.github.sonus21:rqueue-spring-boot-starter:3.0.1-RELEASE' + implementation 'com.github.sonus21:rqueue-spring-boot-starter:3.1.0-RELEASE' ``` * Maven ```xml com.github.sonus21 rqueue-spring-boot-starter - 3.0.1-RELEASE + 3.1.0-RELEASE ``` @@ -140,14 +140,14 @@ in your dependency manager. * Gradle ```groovy - implementation 'com.github.sonus21:rqueue-spring:2.9.0-RELEASE' + implementation 'com.github.sonus21:rqueue-spring:2.13.1-RELEASE' ``` * Maven ```xml com.github.sonus21 rqueue-spring - 2.9.0-RELEASE + 2.13.1-RELEASE ``` @@ -155,14 +155,14 @@ in your dependency manager. * Gradle ```groovy - implementation 'com.github.sonus21:rqueue-spring:3.0.1-RELEASE' + implementation 'com.github.sonus21:rqueue-spring:3.1.0-RELEASE' ``` * Maven ```xml com.github.sonus21 rqueue-spring - 3.0.1-RELEASE + 3.1.0-RELEASE ``` @@ -314,6 +314,8 @@ Rqueue is stable and production ready, it's processing millions of on messages d [![Line](static/users/line.png){: width="70" height="60" alt="Line Chat" }](https://line.me){:target="_blank" style="margin:10px"} [![Aviva](static/users/aviva.jpeg){: width="70" height="60" alt="Aviva" }](https://www.aviva.com/){:target="_blank" style="margin:10px"} [![Diamler Truck](static/users/mercedes.png){: width="80" height="60" alt="Daimler Truck (Mercedes)" }](https://www.daimlertruck.com/en){:target="_blank" style="margin:10px"} +[![Vonage](static/users/vonage.png){: width="250" height="60" alt="Vonage" }](http://vonage.com){:target="_blank" style="margin:10px"} +[![T-Mobile](static/users/t-mobile.svg){: width="50" height="50" alt="T Mobile" }](https://www.t-mobile.com){:target="_blank" style="margin:10px"} [![Poker Stars](static/users/pokerstars.png){: width="210" height="60" alt="PokerStars" }](https://www.pokerstarssports.eu){:target="_blank" style="margin:10px"} [![Tune You](static/users/tuneyou.png){: width="140" height="60" alt="TuneYou" }](https://tuneyou.com){:target="_blank" style="margin:10px"} [![Bit bot](static/users/bitbot.png){: width="80" height="60" alt="BitBot" }](https://bitbot.plus){:target="_blank" style="margin:10px"} diff --git a/docs/message-handling/producer-consumer.md b/docs/message-handling/producer-consumer.md index b6c10b70e..1e0d146b2 100644 --- a/docs/message-handling/producer-consumer.md +++ b/docs/message-handling/producer-consumer.md @@ -163,10 +163,10 @@ supported configurations. setting. * `rqueue.scheduler.auto.start=true` Rqueue scheduler also have thread pools, that handles the message. If you would like to use only event based message movement than set auto start as false. -* `rqueue.scheduler.scheduled.message.thread.pool.size=5` There could be many delayed queues, in that - case Rqueue has to move more messages from ZSET to LIST. In such cases, you can increase thread - pool size, the number of threads used for message movement is minimum of queue count and pool - size. +* `rqueue.scheduler.scheduled.message.thread.pool.size=5` There could be many delayed queues, in + that case Rqueue has to move more messages from ZSET to LIST. In such cases, you can increase + thread pool size, the number of threads used for message movement is minimum of queue count and + pool size. * `rqueue.scheduler.processing.message.thread.pool.size=1` there could be some dead message in processing queue as well, if you're seeing large number of dead messages in processing queue, then you should increase the thread pool size. Processing queue is used for at least once message @@ -174,6 +174,18 @@ supported configurations. * `rqueue.scheduler.scheduled.message.time.interval=5000` At what interval message should be moved from scheduled queue to normal queue. The default value is 5 seconds, that means, you can observe minimum delay of 5 seconds in delayed message consumption. +* `rqueue.scheduler.max.message.count=100` Rqueue continuously move scheduled messages from + processing/scheduled queue to normal queue so that we can process them asap. There are many + instances when large number of messages are scheduled to be run in next 5 minutes. In such cases + Rqueue can pull message from scheduled queue to normal queue at higher rate. By default, it copies + 100 messages from scheduled/processing queue to normal queue. +* `rqueue.scheduler.max.message.mover.delay=60000` Rqueue continuously move scheduled messages from + processing/scheduled queue to normal queue so that we can process them asap. Due to failure, it + can load the Redis system, in such cases it uses exponential backoff to limit the damage. This + time indicates maximum time for which it should wait before making Redis calls. +* `rqueue.scheduler.min.message.mover.delay=200` Rqueue continuously move scheduled messages from + processing/scheduled queue to normal queue so that we can process them asap. It periodically + fetches the messages, the minium delay in such cases can be configured using this variable. ### Dead Letter Queue Consumer/Listener diff --git a/docs/static/users/t-mobile.svg b/docs/static/users/t-mobile.svg new file mode 100644 index 000000000..8881405c7 --- /dev/null +++ b/docs/static/users/t-mobile.svg @@ -0,0 +1,4 @@ + + + + diff --git a/docs/static/users/vonage.png b/docs/static/users/vonage.png new file mode 100644 index 000000000..afd0325be Binary files /dev/null and b/docs/static/users/vonage.png differ diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/config/RqueueSchedulerConfig.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/config/RqueueSchedulerConfig.java index d062e65e8..328831797 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/config/RqueueSchedulerConfig.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/config/RqueueSchedulerConfig.java @@ -16,6 +16,7 @@ package com.github.sonus21.rqueue.config; +import com.github.sonus21.rqueue.utils.Constants; import lombok.Getter; import lombok.Setter; import org.springframework.beans.factory.annotation.Value; @@ -68,8 +69,26 @@ public class RqueueSchedulerConfig { @Value("${rqueue.scheduler.scheduled.message.time.interval:2000}") private long scheduledMessageTimeIntervalInMilli; + // How long the application should wait for task termination + @Value("${rqueue.scheduler.termination.wait.time:200}") + private long terminationWaitTime; // Maximum delay for message mover task due to failure @Value("${rqueue.scheduler.max.message.mover.delay:60000}") private long maxMessageMoverDelay; + + // Minimum amount of time between two consecutive message move calls + @Value("${rqueue.scheduler.min.message.mover.delay:100}") + private long minMessageMoverDelay; + + // Maximum number of messages that should be copied from scheduled to normal queue + @Value("${rqueue.scheduler.max.message.count:100}") + private long maxMessageCount; + + public long minMessageMoveDelay() { + if (minMessageMoverDelay <= 0) { + return Constants.MIN_SCHEDULE_INTERVAL; + } + return minMessageMoverDelay; + } } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/MessageScheduler.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/MessageScheduler.java index e6fe94b73..2998b6add 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/MessageScheduler.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/MessageScheduler.java @@ -16,8 +16,6 @@ package com.github.sonus21.rqueue.core; -import static com.github.sonus21.rqueue.utils.Constants.MAX_MESSAGES; -import static com.github.sonus21.rqueue.utils.Constants.MIN_DELAY; import static java.lang.Math.min; import com.github.sonus21.rqueue.config.RqueueConfig; @@ -27,33 +25,31 @@ import com.github.sonus21.rqueue.models.event.RqueueBootstrapEvent; import com.github.sonus21.rqueue.utils.Constants; import com.github.sonus21.rqueue.utils.ThreadUtils; -import java.time.Instant; +import java.time.Duration; import java.util.Arrays; import java.util.List; import java.util.Map; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; -import lombok.AllArgsConstructor; -import lombok.ToString; +import java.util.concurrent.ScheduledFuture; +import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationListener; import org.springframework.data.redis.RedisSystemException; -import org.springframework.data.redis.connection.Message; -import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.script.DefaultScriptExecutor; import org.springframework.data.redis.core.script.RedisScript; -import org.springframework.data.redis.listener.ChannelTopic; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; -public abstract class MessageScheduler - implements DisposableBean, ApplicationListener { +public abstract class MessageScheduler implements DisposableBean, + ApplicationListener { private final Object monitor = new Object(); @Autowired @@ -61,12 +57,14 @@ public abstract class MessageScheduler @Autowired protected RqueueConfig rqueueConfig; private RedisScript redisScript; - private MessageSchedulerListener messageSchedulerListener; private DefaultScriptExecutor defaultScriptExecutor; private Map queueRunningState; - private Map queueNameToScheduledTask; - private Map channelNameToQueueName; - private Map queueNameToLastMessageScheduleTime; + private Map> queueNameToScheduledTask; + private Map queueNameToNextRunTime; + + @VisibleForTesting + protected RedisScheduleTriggerHandler redisScheduleTriggerHandler; + private ThreadPoolTaskScheduler scheduler; @Autowired private RqueueRedisListenerContainerFactory rqueueRedisListenerContainerFactory; @@ -74,13 +72,11 @@ public abstract class MessageScheduler @Autowired @Qualifier("rqueueRedisLongTemplate") private RedisTemplate redisTemplate; - - private Map queueSchedulers; private Map errorCount; protected abstract Logger getLogger(); - protected abstract long getNextScheduleTime(String queueName, Long value); + protected abstract long getNextScheduleTime(String queueName, long currentTime, Long value); protected abstract String getChannelName(String queueName); @@ -90,7 +86,15 @@ public abstract class MessageScheduler protected abstract int getThreadPoolSize(); - protected abstract boolean isProcessingQueue(String queueName); + protected Duration getPeriod() { + long delay = rqueueSchedulerConfig.getScheduledMessageTimeIntervalInMilli(); + if (delay <= 0) { + delay = Constants.MIN_SCHEDULE_INTERVAL; + } + return Duration.ofMillis(delay); + } + + protected abstract boolean isProcessingQueue(); private void doStart() { for (String queueName : queueRunningState.keySet()) { @@ -98,14 +102,16 @@ private void doStart() { } } - private void subscribeToRedisTopic(String queueName) { - if (isRedisEnabled()) { - String channelName = getChannelName(queueName); - getLogger().debug("Queue {} subscribe to channel {}", queueName, channelName); - this.rqueueRedisListenerContainerFactory.addMessageListener( - messageSchedulerListener, new ChannelTopic(channelName)); - channelNameToQueueName.put(channelName, queueName); - } + private MessageMoverTask task(String queueName, boolean periodic) { + QueueDetail queueDetail = EndpointRegistry.get(queueName); + String zsetName = getZsetName(queueName); + return new MessageMoverTask(queueDetail, zsetName, periodic); + } + + protected void schedule(String queueName) { + Runnable task = task(queueName, true); + ScheduledFuture future = scheduler.scheduleAtFixedRate(task, getPeriod()); + queueNameToScheduledTask.put(queueName, future); } private void startQueue(String queueName) { @@ -113,15 +119,12 @@ private void startQueue(String queueName) { return; } queueRunningState.put(queueName, true); - if (scheduleTaskAtStartup() || !isRedisEnabled()) { - long scheduleAt = getQueueStartTime(); - schedule(queueName, scheduleAt, false); + if (rqueueSchedulerConfig.isAutoStart()) { + schedule(queueName); + } + if (isRedisEnabled()) { + redisScheduleTriggerHandler.startQueue(queueName); } - subscribeToRedisTopic(queueName); - } - - protected long getQueueStartTime() { - return System.currentTimeMillis() + MIN_DELAY; } private void doStop() { @@ -135,33 +138,31 @@ private void doStop() { } waitForRunningQueuesToStop(); queueNameToScheduledTask.clear(); + if (isRedisEnabled()) { + redisScheduleTriggerHandler.stop(); + } } private void waitForRunningQueuesToStop() { for (Map.Entry runningState : queueRunningState.entrySet()) { String queueName = runningState.getKey(); - ScheduledTaskDetail scheduledTaskDetail = queueNameToScheduledTask.get(queueName); - if (scheduledTaskDetail != null) { - Future future = scheduledTaskDetail.getFuture(); - boolean completedOrCancelled = future.isCancelled() || future.isDone(); - if (!completedOrCancelled) { - future.cancel(true); - } - } + ScheduledFuture scheduledFuture = queueNameToScheduledTask.get(queueName); + ThreadUtils.waitForTermination( + getLogger(), + scheduledFuture, + rqueueSchedulerConfig.getTerminationWaitTime(), + "An exception occurred while stopping scheduler queue '{}'", + queueName + ); } } private void stopQueue(String queueName) { - Assert.isTrue( - queueRunningState.containsKey(queueName), + Assert.isTrue(queueRunningState.containsKey(queueName), "Queue with name '" + queueName + "' does not exist"); queueRunningState.put(queueName, false); } - private boolean scheduleTaskAtStartup() { - return rqueueSchedulerConfig.isAutoStart(); - } - private boolean isRedisEnabled() { return rqueueSchedulerConfig.isRedisEnabled(); } @@ -195,40 +196,32 @@ private boolean isQueueActive(String queueName) { return val; } - private long getLastScheduleTime(String queueName) { - return queueNameToLastMessageScheduleTime.getOrDefault(queueName, 0L); - } - - protected ScheduledTaskDetail getScheduledTask(String queueName) { - return queueNameToScheduledTask.get(queueName); - } - - protected void schedule(String queueName, Long startTime, boolean forceSchedule) { - this.queueSchedulers.get(queueName).schedule(queueName, startTime, forceSchedule); - } - protected void initialize() { List queueNames = EndpointRegistry.getActiveQueues(); defaultScriptExecutor = new DefaultScriptExecutor<>(redisTemplate); redisScript = RedisScriptFactory.getScript(ScriptType.MOVE_EXPIRED_MESSAGE); queueRunningState = new ConcurrentHashMap<>(queueNames.size()); queueNameToScheduledTask = new ConcurrentHashMap<>(queueNames.size()); - channelNameToQueueName = new ConcurrentHashMap<>(queueNames.size()); - queueNameToLastMessageScheduleTime = new ConcurrentHashMap<>(queueNames.size()); - queueSchedulers = new ConcurrentHashMap<>(queueNames.size()); + queueNameToNextRunTime = new ConcurrentHashMap<>(queueNames.size()); errorCount = new ConcurrentHashMap<>(queueNames.size()); createScheduler(queueNames.size()); - if (isRedisEnabled()) { - messageSchedulerListener = new MessageSchedulerListener(); - } for (String queueName : queueNames) { initQueue(queueName); } + if (isRedisEnabled()) { + redisScheduleTriggerHandler = new RedisScheduleTriggerHandler( + getLogger(), + rqueueRedisListenerContainerFactory, + rqueueSchedulerConfig, + queueNames, + this::addTask, + this::getChannelName); + redisScheduleTriggerHandler.initialize(); + } } private void initQueue(String queueName) { queueRunningState.put(queueName, false); - queueSchedulers.put(queueName, new QueueScheduler()); } @Override @@ -256,118 +249,29 @@ public void onApplicationEvent(RqueueBootstrapEvent event) { } } - protected long getMinDelay() { - return MIN_DELAY; + protected Future addTask(String queueName) { + return scheduler.submit(task(queueName, false)); } - private class QueueScheduler { - - private void updateLastScheduleTime(String queueName, long time) { - queueNameToLastMessageScheduleTime.put(queueName, time); - } - - private void scheduleNewTask( - QueueDetail queueDetail, String queueName, String zsetName, long startTime) { - MessageMoverTask timerTask = - new MessageMoverTask( - queueDetail.getName(), - queueDetail.getQueueName(), - zsetName, - isProcessingQueue(zsetName)); - Future future = - scheduler.schedule( - timerTask, Instant.ofEpochMilli(getNextScheduleTime(queueName, startTime))); - addTask(timerTask, new ScheduledTaskDetail(startTime, future)); - } - - private void scheduleTask( - long startTime, long currentTime, QueueDetail queueDetail, String zsetName) { - long requiredDelay = Math.max(1, startTime - currentTime); - long taskStartTime = startTime; - MessageMoverTask timerTask = - new MessageMoverTask( - queueDetail.getName(), - queueDetail.getQueueName(), - zsetName, - isProcessingQueue(queueDetail.getName())); - Future future; - if (requiredDelay < MIN_DELAY) { - future = scheduler.submit(timerTask); - taskStartTime = currentTime; - } else { - future = scheduler.schedule(timerTask, Instant.ofEpochMilli(currentTime + requiredDelay)); - } - addTask(timerTask, new ScheduledTaskDetail(taskStartTime, future)); - } - - private boolean shouldNotSchedule(String queueName, boolean forceSchedule) { - boolean isQueueActive = isQueueActive(queueName); - if (!isQueueActive || scheduler == null) { - return true; - } - long lastSeenTime = getLastScheduleTime(queueName); - long currentTime = System.currentTimeMillis(); - // ignore too frequents events - return !forceSchedule && currentTime - lastSeenTime < getMinDelay(); - } - - protected synchronized void schedule(String queueName, Long startTime, boolean forceSchedule) { - if (shouldNotSchedule(queueName, forceSchedule)) { - return; - } - long currentTime = System.currentTimeMillis(); - updateLastScheduleTime(queueName, currentTime); - ScheduledTaskDetail scheduledTaskDetail = getScheduledTask(queueName); - QueueDetail queueDetail = EndpointRegistry.get(queueName); - String zsetName = getZsetName(queueName); - if (scheduledTaskDetail == null || forceSchedule) { - scheduleTask(startTime, currentTime, queueDetail, zsetName); - return; - } - checkExistingTask(scheduledTaskDetail, currentTime, queueDetail, zsetName); - scheduleNewTask(queueDetail, queueName, zsetName, startTime); - } - - private void addTask(MessageMoverTask timerTask, ScheduledTaskDetail scheduledTaskDetail) { - getLogger().debug("Timer: {}, Task: {}", timerTask, scheduledTaskDetail); - queueNameToScheduledTask.put(timerTask.getName(), scheduledTaskDetail); - } - - private void checkExistingTask( - ScheduledTaskDetail scheduledTaskDetail, - long currentTime, - QueueDetail queueDetail, - String zsetName) { - // run existing tasks continue - long existingDelay = scheduledTaskDetail.getStartTime() - currentTime; - Future submittedTask = scheduledTaskDetail.getFuture(); - boolean completedOrCancelled = submittedTask.isDone() || submittedTask.isCancelled(); - // tasks older than TASK_ALIVE_TIME are considered dead - if (!completedOrCancelled - && existingDelay < MIN_DELAY - && existingDelay > Constants.TASK_ALIVE_TIME) { - ThreadUtils.waitForTermination( - getLogger(), - submittedTask, - Constants.DEFAULT_SCRIPT_EXECUTION_TIME, - "LIST: {} ZSET: {}, Task: {} failed", - queueDetail.getQueueName(), - zsetName, - scheduledTaskDetail); - } - } - } - - @ToString - @AllArgsConstructor private class MessageMoverTask implements Runnable { + private final String id; private final String name; private final String queueName; private final String zsetName; private final boolean processingQueue; + private final boolean periodic; + + MessageMoverTask(QueueDetail queueDetail, String zsetName, boolean periodic) { + this.id = UUID.randomUUID().toString(); + this.name = queueDetail.getName(); + this.queueName = queueDetail.getQueueName(); + this.zsetName = zsetName; + this.periodic = periodic; + this.processingQueue = isProcessingQueue(); + } - private long getNextScheduleTimeInternal(Long value, Exception e) { + private long getNextScheduleTimeInternal(Long value, long currentTime, Exception e) { int errCount = 0; long nextTime; if (null != e) { @@ -375,31 +279,53 @@ private long getNextScheduleTimeInternal(Long value, Exception e) { if (errCount % 3 == 0) { getLogger().error("Message mover task is failing continuously queue: {}", name, e); } - long delay = (long) (100 * Math.pow(1.5, errCount)); - delay = Math.min(delay, rqueueSchedulerConfig.getMaxMessageMoverDelay()); - nextTime = System.currentTimeMillis() + delay; + // delay = x * 1.5^errorCount + double delay = rqueueSchedulerConfig.minMessageMoveDelay() * Math.pow(1.5, errCount); + long maxDelay = Math.min((long) delay, rqueueSchedulerConfig.getMaxMessageMoverDelay()); + nextTime = currentTime + maxDelay; } else { - nextTime = getNextScheduleTime(name, value); + nextTime = getNextScheduleTime(name, currentTime, value); } errorCount.put(name, errCount); return nextTime; } + @Override + public String toString() { + return String.format("MessageMoverTask(id=%s, queue=%s, periodic=%s)", id, name, periodic); + } + + private long getMessageCount() { + return rqueueSchedulerConfig.getMaxMessageCount(); + } + + private List scriptKeys() { + return Arrays.asList(queueName, zsetName); + } + + private Object[] scriptArgs() { + long currentTime = System.currentTimeMillis(); + return new Object[]{currentTime, getMessageCount(), processingQueue ? 1 : 0}; + } + + private boolean shouldSkip(long currentTime) { + Long nextRunTime = queueNameToNextRunTime.get(queueName); + return nextRunTime != null && nextRunTime > currentTime; + } + @Override public void run() { + long currentTime = System.currentTimeMillis(); + if (shouldSkip(currentTime)) { + getLogger().debug("Skipped {}", this); + return; + } getLogger().debug("Running {}", this); Long value = null; Exception e = null; try { if (isQueueActive(name)) { - long currentTime = System.currentTimeMillis(); - value = - defaultScriptExecutor.execute( - redisScript, - Arrays.asList(queueName, zsetName), - currentTime, - MAX_MESSAGES, - processingQueue ? 1 : 0); + value = defaultScriptExecutor.execute(redisScript, scriptKeys(), scriptArgs()); } } catch (RedisSystemException ex) { e = ex; @@ -407,10 +333,8 @@ public void run() { e = ex; getLogger().warn("Task execution failed for the queue: {}", getName(), e); } finally { - if (isQueueActive(name)) { - long nextExecutionTime = getNextScheduleTimeInternal(value, e); - schedule(name, nextExecutionTime, true); - } + long nextExecutionTime = getNextScheduleTimeInternal(value, currentTime, e); + queueNameToNextRunTime.put(queueName, nextExecutionTime); } } @@ -419,36 +343,4 @@ public String getName() { } } - private class MessageSchedulerListener implements MessageListener { - - private void handleMessage(String queueName, Long startTime) { - long lastSeenTime = getLastScheduleTime(queueName); - long currentTime = System.currentTimeMillis(); - if (currentTime - lastSeenTime < getMinDelay()) { - return; - } - schedule(queueName, startTime, false); - } - - @Override - public void onMessage(Message message, byte[] pattern) { - if (message.getBody().length == 0 || message.getChannel().length == 0) { - return; - } - String body = new String(message.getBody()); - String channel = new String(message.getChannel()); - getLogger().trace("Body: {} Channel: {}", body, channel); - try { - Long startTime = Long.parseLong(body); - String queueName = channelNameToQueueName.get(channel); - if (queueName == null) { - getLogger().warn("Unknown channel name {}", channel); - return; - } - handleMessage(queueName, startTime); - } catch (Exception e) { - getLogger().error("Error occurred on a channel {}, body: {}", channel, body, e); - } - } - } } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ProcessingQueueMessageScheduler.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ProcessingQueueMessageScheduler.java index da1420749..276a09032 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ProcessingQueueMessageScheduler.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ProcessingQueueMessageScheduler.java @@ -19,9 +19,11 @@ import static java.lang.Long.max; import com.github.sonus21.rqueue.listener.QueueDetail; +import java.time.Duration; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import com.github.sonus21.rqueue.utils.Constants; import lombok.extern.slf4j.Slf4j; import org.slf4j.Logger; @@ -61,7 +63,7 @@ protected int getThreadPoolSize() { } @Override - protected boolean isProcessingQueue(String queueName) { + protected boolean isProcessingQueue() { return true; } @@ -71,8 +73,7 @@ protected String getThreadNamePrefix() { } @Override - protected long getNextScheduleTime(String queueName, Long value) { - long currentTime = System.currentTimeMillis(); + protected long getNextScheduleTime(String queueName, long currentTime, Long value) { if (value == null) { long delay = queueNameToDelay.get(queueName); return currentTime + delay; diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RedisScheduleTriggerHandler.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RedisScheduleTriggerHandler.java new file mode 100644 index 000000000..4fb41d1b8 --- /dev/null +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RedisScheduleTriggerHandler.java @@ -0,0 +1,161 @@ +/* + * Copyright (c) 2019-2023 Sonu Kumar + * + * 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 + * + * https://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 com.github.sonus21.rqueue.core; + +import com.github.sonus21.rqueue.config.RqueueSchedulerConfig; +import com.github.sonus21.rqueue.utils.ThreadUtils; +import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.listener.ChannelTopic; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Future; +import java.util.function.Function; + +class RedisScheduleTriggerHandler { + + private final RqueueRedisListenerContainerFactory rqueueRedisListenerContainerFactory; + private final RqueueSchedulerConfig rqueueSchedulerConfig; + private final Logger logger; + private final Function> scheduler; + private final Function channelNameProducer; + private final List queueNames; + + @VisibleForTesting + Map queueNameToLastRunTime; + @VisibleForTesting + Map> queueNameToFuture; + @VisibleForTesting + Map channelNameToQueueName; + @VisibleForTesting + MessageListener messageListener; + + RedisScheduleTriggerHandler(Logger logger, + RqueueRedisListenerContainerFactory rqueueRedisListenerContainerFactory, + RqueueSchedulerConfig rqueueSchedulerConfig, List queueNames, + Function> scheduler, Function channelNameProducer) { + this.queueNames = queueNames; + this.rqueueSchedulerConfig = rqueueSchedulerConfig; + this.rqueueRedisListenerContainerFactory = rqueueRedisListenerContainerFactory; + this.logger = logger; + this.scheduler = scheduler; + this.channelNameProducer = channelNameProducer; + } + + void initialize() { + this.messageListener = new MessageSchedulerListener(); + this.channelNameToQueueName = new HashMap<>(queueNames.size()); + this.queueNameToFuture = new ConcurrentHashMap<>(queueNames.size()); + this.queueNameToLastRunTime = new ConcurrentHashMap<>(queueNames.size()); + } + + void stop() { + for (String queue : queueNames) { + stopQueue(queue); + } + } + + void startQueue(String queueName) { + queueNameToLastRunTime.put(queueName, 0L); + subscribeToRedisTopic(queueName); + } + + void stopQueue(String queueName) { + Future future = queueNameToFuture.get(queueName); + ThreadUtils.waitForTermination(logger, future, rqueueSchedulerConfig.getTerminationWaitTime(), + "An exception occurred while stopping scheduler queue '{}'", queueName); + queueNameToLastRunTime.put(queueName, 0L); + queueNameToFuture.remove(queueName); + unsubscribeFromRedis(queueName); + } + + private void unsubscribeFromRedis(String queueName) { + String channelName = channelNameProducer.apply(queueName); + logger.debug("Queue {} unsubscribe from channel {}", queueName, channelName); + rqueueRedisListenerContainerFactory.removeMessageListener(messageListener, + new ChannelTopic(channelName)); + channelNameToQueueName.put(channelName, queueName); + } + + private void subscribeToRedisTopic(String queueName) { + String channelName = channelNameProducer.apply(queueName); + channelNameToQueueName.put(channelName, queueName); + logger.debug("Queue {} subscribe to channel {}", queueName, channelName); + rqueueRedisListenerContainerFactory.addMessageListener(messageListener, + new ChannelTopic(channelName)); + } + + protected long getMinDelay() { + return rqueueSchedulerConfig.minMessageMoveDelay(); + } + + + /** + * This MessageListener listen the event from Redis, its expected that the event should be only + * raised when elements in the ZSET are lagging behind current time. + */ + private class MessageSchedulerListener implements MessageListener { + + private void schedule(String queueName, long currentTime) { + Future future = queueNameToFuture.get(queueName); + if (future == null || future.isCancelled() || future.isDone()) { + queueNameToLastRunTime.put(queueName, currentTime); + Future newFuture = scheduler.apply(queueName); + queueNameToFuture.put(queueName, newFuture); + } + } + + private void handleMessage(String queueName, long startTime) { + long currentTime = System.currentTimeMillis(); + if (startTime > currentTime) { + logger.warn("Received message body is not correct queue: {}, time: {}", queueName, + startTime); + return; + } + long lastRunTime = queueNameToLastRunTime.get(queueName); + if (currentTime - lastRunTime < getMinDelay()) { + return; + } + schedule(queueName, currentTime); + } + + @Override + public void onMessage(Message message, byte[] pattern) { + if (message.getBody().length == 0 || message.getChannel().length == 0) { + return; + } + String body = new String(message.getBody()); + String channel = new String(message.getChannel()); + logger.trace("Body: {} Channel: {}", body, channel); + try { + long startTime = Long.parseLong(body); + String queueName = channelNameToQueueName.get(channel); + if (queueName == null) { + logger.warn("Unknown channel name {}", channel); + return; + } + handleMessage(queueName, startTime); + } catch (Exception e) { + logger.error("Error occurred on a channel {}, body: {}", channel, body, e); + } + } + } +} diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueRedisListenerContainerFactory.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueRedisListenerContainerFactory.java index 95215d3d2..1b6cf3edb 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueRedisListenerContainerFactory.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/RqueueRedisListenerContainerFactory.java @@ -24,6 +24,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.SmartLifecycle; import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.listener.ChannelTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.listener.Topic; @@ -96,7 +97,7 @@ public void afterPropertiesSet() throws Exception { createContainer(); return; } - if (rqueueConfig.isSharedConnection() || rqueueSchedulerConfig.isListenerShared()) { + if (rqueueSchedulerConfig.isListenerShared()) { if (systemContainer != null) { container = systemContainer; sharedContainer = true; @@ -121,4 +122,8 @@ public void stop(Runnable callback) { public int getPhase() { return Integer.MAX_VALUE; } + + public void removeMessageListener(MessageListener messageListener, ChannelTopic channelTopic) { + getContainer().removeMessageListener(messageListener, channelTopic); + } } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ScheduledQueueMessageScheduler.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ScheduledQueueMessageScheduler.java index 91b815230..ac660b091 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ScheduledQueueMessageScheduler.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ScheduledQueueMessageScheduler.java @@ -28,8 +28,7 @@ protected Logger getLogger() { } @Override - protected long getNextScheduleTime(String queueName, Long value) { - long currentTime = System.currentTimeMillis(); + protected long getNextScheduleTime(String queueName, long currentTime, Long value) { if (value == null) { return currentTime + rqueueSchedulerConfig.getScheduledMessageTimeIntervalInMilli(); } @@ -60,7 +59,7 @@ protected int getThreadPoolSize() { } @Override - protected boolean isProcessingQueue(String queueName) { + protected boolean isProcessingQueue() { return false; } } diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ScheduledTaskDetail.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ScheduledTaskDetail.java deleted file mode 100644 index 32e7057f7..000000000 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/core/ScheduledTaskDetail.java +++ /dev/null @@ -1,57 +0,0 @@ -/* - * Copyright (c) 2019-2023 Sonu Kumar - * - * 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 - * - * https://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 com.github.sonus21.rqueue.core; - -import java.util.UUID; -import java.util.concurrent.Future; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.TimeUnit; -import lombok.Getter; - -@Getter -class ScheduledTaskDetail { - - private final Future future; - private final long startTime; - private final String id; - - ScheduledTaskDetail(long startTime, Future future) { - this.startTime = startTime; - this.future = future; - this.id = UUID.randomUUID().toString(); - } - - @Override - public String toString() { - StringBuilder sb = new StringBuilder(); - if (future instanceof ScheduledFuture) { - sb.append("ScheduledFuture(delay="); - sb.append(((ScheduledFuture) future).getDelay(TimeUnit.MILLISECONDS)); - sb.append("Ms, "); - } else { - sb.append("Future("); - } - sb.append("id="); - sb.append(id); - sb.append(", startTime="); - sb.append(startTime); - sb.append(", currentTime="); - sb.append(System.currentTimeMillis()); - sb.append(")"); - return sb.toString(); - } -} diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/Constants.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/Constants.java index 65e8449e9..12fee1798 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/Constants.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/Constants.java @@ -32,12 +32,11 @@ public final class Constants { public static final int MINUTES_IN_A_DAY = HOURS_IN_A_DAY * MINUTES_IN_AN_HOUR; public static final int SECONDS_IN_A_DAY = MINUTES_IN_A_DAY * SECONDS_IN_A_MINUTE; public static final long MILLIS_IN_A_DAY = SECONDS_IN_A_DAY * ONE_MILLI; - public static final int SECONDS_IN_A_WEEK = DAYS_IN_A_WEEK * SECONDS_IN_A_DAY; public static final long DEFAULT_SCRIPT_EXECUTION_TIME = 5 * ONE_MILLI; public static final long MIN_DELAY = 100L; + public static final long MIN_SCHEDULE_INTERVAL = 100L; public static final long MIN_EXECUTION_TIME = MIN_DELAY; public static final long DELTA_BETWEEN_RE_ENQUEUE_TIME = ONE_MILLI; - public static final long TASK_ALIVE_TIME = -30 * ONE_MILLI; public static final int DEFAULT_RETRY_DEAD_LETTER_QUEUE = 3; public static final int MAX_MESSAGES = 100; public static final int DEFAULT_WORKER_COUNT_PER_QUEUE = 2; diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/ThreadUtils.java b/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/ThreadUtils.java index a033dc69d..45cff7067 100644 --- a/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/ThreadUtils.java +++ b/rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/ThreadUtils.java @@ -78,16 +78,14 @@ private static void waitForShutdown( public static void waitForTermination( Logger log, Future future, long waitTimeInMillis, String msg, Object... msgParams) { - if (future == null) { + if (future == null || future.isCancelled() || future.isDone()) { return; } - boolean completedOrCancelled = future.isCancelled() || future.isDone(); - if (!completedOrCancelled) { - if (future instanceof ScheduledFuture) { - ScheduledFuture f = (ScheduledFuture) future; - if (f.getDelay(TimeUnit.MILLISECONDS) > Constants.MIN_DELAY) { - return; - } + if (future instanceof ScheduledFuture) { + ScheduledFuture f = (ScheduledFuture) future; + if (f.getDelay(TimeUnit.MILLISECONDS) > Constants.MIN_DELAY) { + f.cancel(false); + return; } } waitForShutdown(log, future, waitTimeInMillis, msg, msgParams); diff --git a/rqueue-core/src/main/resources/META-INF/spring-configuration-metadata.json b/rqueue-core/src/main/resources/META-INF/spring-configuration-metadata.json index 3cc46a5eb..7f7c484b2 100644 --- a/rqueue-core/src/main/resources/META-INF/spring-configuration-metadata.json +++ b/rqueue-core/src/main/resources/META-INF/spring-configuration-metadata.json @@ -263,14 +263,14 @@ { "sourceType" : "com.github.sonus21.rqueue.config.RqueueSchedulerConfig", "name" : "rqueue.scheduler.redis.enabled", - "description" : "Scheduler Should auto start or not", + "description" : "Scheduler Should use redis PUB/SUB", "type" : "java.lang.Boolean", "defaultValue" : true }, { "sourceType" : "com.github.sonus21.rqueue.config.RqueueSchedulerConfig", "name" : "rqueue.scheduler.scheduled.message.thread.pool.size", - "description" : "Number of threads should be used to pull scheduled message to original queue", + "description" : "Number of threads that will be used to pull scheduled message to normal queue", "type" : "java.lang.Integer", "defaultValue" : 3 }, @@ -281,6 +281,13 @@ "type" : "java.lang.Integer", "defaultValue" : 1 }, + { + "sourceType" : "com.github.sonus21.rqueue.config.RqueueSchedulerConfig", + "name" : "rqueue.scheduler.max.message.count", + "description" : "Maximum number of messages that should be moved to normal queue from processing/schedule queue", + "type" : "java.lang.Integer", + "defaultValue" : 100 + }, { "sourceType" : "com.github.sonus21.rqueue.config.RqueueSchedulerConfig", "name" : "rqueue.scheduler.scheduled.message.time.interval", @@ -288,6 +295,28 @@ "type" : "java.lang.Long", "defaultValue" : 5000 }, + { + "sourceType" : "com.github.sonus21.rqueue.config.RqueueSchedulerConfig", + "name" : "rqueue.scheduler.termination.wait.time", + "description" : "How long the application would wait for task termination", + "type" : "java.lang.Long", + "defaultValue" : 200 + }, + + { + "sourceType" : "com.github.sonus21.rqueue.config.RqueueSchedulerConfig", + "name" : "rqueue.scheduler.max.message.mover.delay", + "description" : "Maximum delay for message mover task due to failure", + "type" : "java.lang.Long", + "defaultValue" : 60000 + }, + { + "sourceType" : "com.github.sonus21.rqueue.config.RqueueSchedulerConfig", + "name" : "rqueue.scheduler.min.message.mover.delay", + "description" : "Minimum amount of time between two consecutive message move calls", + "type" : "java.lang.Long", + "defaultValue" : 100 + }, { "sourceType" : "com.github.sonus21.rqueue.config.RqueueWebConfig", "name" : "rqueue.web.enable", diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulerDisabledTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulerDisabledTest.java index 162fe17ed..6f29bb08c 100644 --- a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulerDisabledTest.java +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulerDisabledTest.java @@ -66,6 +66,7 @@ public void init() { void startShouldSubmitsTaskWhenRedisIsDisabled() throws Exception { doReturn(1).when(rqueueSchedulerConfig).getScheduledMessageThreadPoolSize(); doReturn(true).when(rqueueSchedulerConfig).isEnabled(); + doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); TestTaskScheduler scheduler = new TestTaskScheduler(); try (MockedStatic threadUtils = Mockito.mockStatic(ThreadUtils.class)) { threadUtils @@ -73,7 +74,7 @@ void startShouldSubmitsTaskWhenRedisIsDisabled() throws Exception { .thenReturn(scheduler); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); assertEquals(1, scheduler.submittedTasks()); - assertNull(FieldUtils.readField(messageScheduler, "messageSchedulerListener", true)); + assertNull(messageScheduler.redisScheduleTriggerHandler); messageScheduler.destroy(); } } @@ -86,8 +87,7 @@ void afterPropertiesSetProducerMode() throws Exception { assertNull(FieldUtils.readField(messageScheduler, "scheduler", true)); assertNull(FieldUtils.readField(messageScheduler, "queueRunningState", true)); assertNull(FieldUtils.readField(messageScheduler, "queueNameToScheduledTask", true)); - assertNull(FieldUtils.readField(messageScheduler, "channelNameToQueueName", true)); - assertNull(FieldUtils.readField(messageScheduler, "queueNameToLastMessageScheduleTime", true)); + assertNull(FieldUtils.readField(messageScheduler, "queueNameToNextRunTime", true)); } @Test @@ -112,7 +112,6 @@ void afterPropertiesSetDisabled() throws Exception { assertNull(FieldUtils.readField(messageScheduler, "scheduler", true)); assertNull(FieldUtils.readField(messageScheduler, "queueRunningState", true)); assertNull(FieldUtils.readField(messageScheduler, "queueNameToScheduledTask", true)); - assertNull(FieldUtils.readField(messageScheduler, "channelNameToQueueName", true)); - assertNull(FieldUtils.readField(messageScheduler, "queueNameToLastMessageScheduleTime", true)); + assertNull(FieldUtils.readField(messageScheduler, "queueNameToNextRunTime", true)); } } diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulerTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulerTest.java index 8b52ce9a9..bafefd9a0 100644 --- a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulerTest.java +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulerTest.java @@ -23,7 +23,7 @@ import com.github.sonus21.rqueue.CoreUnitTest; import com.github.sonus21.rqueue.config.RqueueConfig; import com.github.sonus21.rqueue.config.RqueueSchedulerConfig; -import com.github.sonus21.rqueue.core.ScheduledQueueMessageSchedulerTest.TestMessageScheduler; +import com.github.sonus21.rqueue.core.ScheduledQueueMessageSchedulerTest.TestScheduledQueueMessageScheduler; import com.github.sonus21.rqueue.listener.QueueDetail; import com.github.sonus21.rqueue.models.event.RqueueBootstrapEvent; import com.github.sonus21.rqueue.utils.TestUtils; @@ -56,7 +56,7 @@ class MessageSchedulerTest extends TestBase { @Mock private RedisTemplate redisTemplate; @InjectMocks - private TestMessageScheduler messageScheduler; + private TestScheduledQueueMessageScheduler messageScheduler; @BeforeEach public void init() { @@ -69,7 +69,7 @@ public void init() { void afterPropertiesSetWithEmptyQueSet() throws Exception { EndpointRegistry.delete(); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); - assertEquals(0, messageScheduler.scheduleList.size()); + assertEquals(0, messageScheduler.scheduleCounter.get()); messageScheduler.destroy(); } diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageScheduleTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulingTest.java similarity index 62% rename from rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageScheduleTest.java rename to rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulingTest.java index f26a75952..e5bf012cb 100644 --- a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageScheduleTest.java +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/MessageSchedulingTest.java @@ -19,6 +19,7 @@ import static com.github.sonus21.rqueue.utils.TimeoutUtils.sleep; import static com.github.sonus21.rqueue.utils.TimeoutUtils.waitFor; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; @@ -31,9 +32,12 @@ import com.github.sonus21.rqueue.core.ProcessingQueueMessageSchedulerTest.ProcessingQTestMessageScheduler; import com.github.sonus21.rqueue.listener.QueueDetail; import com.github.sonus21.rqueue.models.event.RqueueBootstrapEvent; +import com.github.sonus21.rqueue.utils.Constants; import com.github.sonus21.rqueue.utils.TestUtils; import com.github.sonus21.rqueue.utils.ThreadUtils; +import com.github.sonus21.rqueue.utils.TimeoutUtils; import com.github.sonus21.test.TestTaskScheduler; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -46,12 +50,14 @@ import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.TooManyClusterRedirectionsException; +import org.springframework.data.redis.connection.DefaultMessage; +import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.RedisTemplate; @CoreUnitTest @SuppressWarnings("unchecked") -class MessageScheduleTest extends TestBase { +class MessageSchedulingTest extends TestBase { @InjectMocks private final ProcessingQTestMessageScheduler messageScheduler = new ProcessingQTestMessageScheduler(); @@ -71,100 +77,60 @@ public void init() { MockitoAnnotations.openMocks(this); EndpointRegistry.delete(); EndpointRegistry.register(queueDetail); - } - - @Test - void onCompletionOfExistingTaskNewTaskShouldBeSubmitted() throws Exception { - try (MockedStatic threadUtils = Mockito.mockStatic(ThreadUtils.class)) { - doReturn(1).when(rqueueSchedulerConfig).getProcessingMessageThreadPoolSize(); - doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); - doReturn(true).when(rqueueSchedulerConfig).isEnabled(); - doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); - AtomicInteger counter = new AtomicInteger(0); - doAnswer( - invocation -> { - counter.incrementAndGet(); - return null; - }) - .when(redisTemplate) - .execute(any(RedisCallback.class)); - TestTaskScheduler scheduler = new TestTaskScheduler(); - threadUtils - .when(() -> ThreadUtils.createTaskScheduler(1, "processingQueueMsgScheduler-", 60)) - .thenReturn(scheduler); - messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); - waitFor(() -> counter.get() >= 1, "scripts are getting executed"); - sleep(10); - messageScheduler.destroy(); - assertTrue(scheduler.submittedTasks() >= 2); - } + doReturn(1).when(rqueueSchedulerConfig).getProcessingMessageThreadPoolSize(); + doReturn(200L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli(); + doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); + doReturn(true).when(rqueueSchedulerConfig).isEnabled(); + doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); } @Test void multipleTasksAreRunningForTheSameQueue() throws Exception { try (MockedStatic threadUtils = Mockito.mockStatic(ThreadUtils.class)) { - doReturn(1).when(rqueueSchedulerConfig).getProcessingMessageThreadPoolSize(); - doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); - doReturn(true).when(rqueueSchedulerConfig).isEnabled(); - doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); AtomicInteger counter = new AtomicInteger(0); - doAnswer( - invocation -> { - counter.incrementAndGet(); - return System.currentTimeMillis(); - }) - .when(redisTemplate) - .execute(any(RedisCallback.class)); + doAnswer(invocation -> { + counter.incrementAndGet(); + return System.currentTimeMillis(); + }).when(redisTemplate).execute(any(RedisCallback.class)); TestTaskScheduler scheduler = new TestTaskScheduler(); - threadUtils - .when(() -> ThreadUtils.createTaskScheduler(1, "processingQueueMsgScheduler-", 60)) + threadUtils.when(() -> ThreadUtils.createTaskScheduler(1, "processingQueueMsgScheduler-", 60)) .thenReturn(scheduler); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); waitFor(() -> counter.get() >= 2, "scripts are getting executed"); sleep(10); messageScheduler.destroy(); - assertTrue(scheduler.submittedTasks() >= 3); + assertEquals(1, scheduler.submittedTasks()); } } @Test void taskShouldBeScheduledOnFailure() throws Exception { try (MockedStatic threadUtils = Mockito.mockStatic(ThreadUtils.class)) { - doReturn(1).when(rqueueSchedulerConfig).getProcessingMessageThreadPoolSize(); - doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); - doReturn(true).when(rqueueSchedulerConfig).isEnabled(); - doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); doReturn(10000L).when(rqueueSchedulerConfig).getMaxMessageMoverDelay(); + doReturn(100L).when(rqueueSchedulerConfig).minMessageMoveDelay(); AtomicInteger counter = new AtomicInteger(0); - doAnswer( - invocation -> { - counter.incrementAndGet(); - throw new RedisSystemException("Something is not correct", - new NullPointerException("oops!")); - }) - .when(redisTemplate) - .execute(any(RedisCallback.class)); + doAnswer(invocation -> { + counter.incrementAndGet(); + throw new RedisSystemException("Something is not correct", + new NullPointerException("oops!")); + }).when(redisTemplate).execute(any(RedisCallback.class)); TestTaskScheduler scheduler = new TestTaskScheduler(); - threadUtils - .when(() -> ThreadUtils.createTaskScheduler(1, "processingQueueMsgScheduler-", 60)) + threadUtils.when(() -> ThreadUtils.createTaskScheduler(1, "processingQueueMsgScheduler-", 60)) .thenReturn(scheduler); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); - waitFor(() -> counter.get() >= 2, "scripts are getting executed"); + waitFor(() -> counter.get() >= 3, "scripts are getting executed"); sleep(10); messageScheduler.destroy(); - assertTrue(scheduler.submittedTasks() >= 3); + assertEquals(1, scheduler.submittedTasks()); } } @Test - void continuousTaskFailTask() throws Exception { + void continuousTaskFailure() throws Exception { try (MockedStatic threadUtils = Mockito.mockStatic(ThreadUtils.class)) { - doReturn(1).when(rqueueSchedulerConfig).getProcessingMessageThreadPoolSize(); - doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); - doReturn(true).when(rqueueSchedulerConfig).isEnabled(); - doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); - doReturn(100L).when(rqueueSchedulerConfig).getMaxMessageMoverDelay(); + doReturn(500L).when(rqueueSchedulerConfig).getMaxMessageMoverDelay(); + doReturn(100L).when(rqueueSchedulerConfig).minMessageMoveDelay(); AtomicInteger counter = new AtomicInteger(0); doAnswer( invocation -> { @@ -189,7 +155,7 @@ void continuousTaskFailTask() throws Exception { waitFor(() -> counter.get() >= 5, "scripts are getting executed"); sleep(10); messageScheduler.destroy(); - assertTrue(scheduler.submittedTasks() >= 6); + assertEquals(1, scheduler.submittedTasks()); } } diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/ProcessingQueueMessageSchedulerTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/ProcessingQueueMessageSchedulerTest.java index fbf816f6c..da1d9035e 100644 --- a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/ProcessingQueueMessageSchedulerTest.java +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/ProcessingQueueMessageSchedulerTest.java @@ -26,8 +26,8 @@ import com.github.sonus21.rqueue.config.RqueueSchedulerConfig; import com.github.sonus21.rqueue.listener.QueueDetail; import com.github.sonus21.rqueue.utils.TestUtils; -import java.util.List; -import java.util.Vector; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.InjectMocks; @@ -64,8 +64,7 @@ public void init() { @Test void getChannelName() { - assertEquals( - slowQueueDetail.getProcessingQueueChannelName(), + assertEquals(slowQueueDetail.getProcessingQueueChannelName(), messageScheduler.getChannelName(slowQueue)); } @@ -77,36 +76,41 @@ void getZsetName() { @Test void getNextScheduleTimeSlowQueue() { long currentTime = System.currentTimeMillis(); - assertThat( - messageScheduler.getNextScheduleTime(slowQueue, null), + assertThat(messageScheduler.getNextScheduleTime(slowQueue, currentTime, null), greaterThanOrEqualTo(currentTime + 100000)); - assertEquals( - currentTime + 1000L, messageScheduler.getNextScheduleTime(slowQueue, currentTime + 1000L)); + assertEquals(currentTime + 1000L, + messageScheduler.getNextScheduleTime(slowQueue, currentTime, currentTime + 1000L)); } @Test void getNextScheduleTimeFastQueue() { long currentTime = System.currentTimeMillis(); - assertThat( - messageScheduler.getNextScheduleTime(fastQueue, null), + assertThat(messageScheduler.getNextScheduleTime(fastQueue, currentTime, null), greaterThanOrEqualTo(currentTime + 200000)); - assertEquals( - currentTime + 1000L, messageScheduler.getNextScheduleTime(fastQueue, currentTime + 1000L)); + assertEquals(currentTime + 1000L, + messageScheduler.getNextScheduleTime(fastQueue, currentTime, currentTime + 1000L)); } static class ProcessingQTestMessageScheduler extends ProcessingQueueMessageScheduler { - List scheduleList; + private final AtomicInteger schedulesCalls; + private final AtomicInteger addTaskCalls; ProcessingQTestMessageScheduler() { - this.scheduleList = new Vector<>(); + this.schedulesCalls = new AtomicInteger(0); + this.addTaskCalls = new AtomicInteger(0); } @Override - protected synchronized void schedule(String queueName, Long startTime, boolean forceSchedule) { - super.schedule(queueName, startTime, forceSchedule); - this.scheduleList.add(forceSchedule); + protected synchronized void schedule(String queueName) { + schedulesCalls.incrementAndGet(); + super.schedule(queueName); + } + + protected Future addTask(String queueName) { + addTaskCalls.incrementAndGet(); + return super.addTask(queueName); } } } diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/RedisAndNormalSchedulingTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/RedisAndNormalSchedulingTest.java new file mode 100644 index 000000000..684e9ed9b --- /dev/null +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/RedisAndNormalSchedulingTest.java @@ -0,0 +1,123 @@ +package com.github.sonus21.rqueue.core; + +import com.github.sonus21.TestBase; +import com.github.sonus21.rqueue.CoreUnitTest; +import com.github.sonus21.rqueue.config.RqueueConfig; +import com.github.sonus21.rqueue.config.RqueueSchedulerConfig; +import com.github.sonus21.rqueue.core.ScheduledQueueMessageSchedulerTest.TestScheduledQueueMessageScheduler; +import com.github.sonus21.rqueue.listener.QueueDetail; +import com.github.sonus21.rqueue.models.event.RqueueBootstrapEvent; +import com.github.sonus21.rqueue.utils.Constants; +import com.github.sonus21.rqueue.utils.TestUtils; +import com.github.sonus21.rqueue.utils.ThreadUtils; +import com.github.sonus21.rqueue.utils.TimeoutUtils; +import com.github.sonus21.test.TestTaskScheduler; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.MockitoAnnotations; +import org.springframework.data.redis.connection.DefaultMessage; +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.RedisTemplate; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static com.github.sonus21.rqueue.utils.TimeoutUtils.sleep; +import static com.github.sonus21.rqueue.utils.TimeoutUtils.waitFor; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; + +@CoreUnitTest +@Slf4j +class RedisAndNormalSchedulingTest extends TestBase { + + private final String slowQueue = "slow-queue"; + private final String fastQueue = "fast-queue"; + private final QueueDetail slowQueueDetail = TestUtils.createQueueDetail(slowQueue); + private final QueueDetail fastQueueDetail = TestUtils.createQueueDetail(fastQueue); + @Mock + private RqueueSchedulerConfig rqueueSchedulerConfig; + @Mock + private RqueueConfig rqueueConfig; + @Mock + private RedisTemplate redisTemplate; + @Mock + private RqueueRedisListenerContainerFactory rqueueRedisListenerContainerFactory; + @InjectMocks + private TestScheduledQueueMessageScheduler messageScheduler; + + @BeforeEach + public void init() { + MockitoAnnotations.openMocks(this); + EndpointRegistry.delete(); + EndpointRegistry.register(fastQueueDetail); + EndpointRegistry.register(slowQueueDetail); + } + + @Test + void redisAndNormalScheduling() throws Exception { + TestTaskScheduler scheduler = new TestTaskScheduler(2); + AtomicBoolean startGenerateMessage = new AtomicBoolean(false); + AtomicBoolean generateMessage = new AtomicBoolean(true); + long totalTime = 2000L; + long minDelay = 10L; + //25% buffer due to short polling intervals, IN CI it runs slowly + double buffer = 0.25; + String channelName = messageScheduler.getChannelName(slowQueue); + doReturn(1).when(rqueueSchedulerConfig).getScheduledMessageThreadPoolSize(); + doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); + doReturn(true).when(rqueueSchedulerConfig).isEnabled(); + doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); + doReturn(3000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli(); + doReturn(100L).when(rqueueSchedulerConfig).getMaxMessageCount(); + doReturn(minDelay).when(rqueueSchedulerConfig).minMessageMoveDelay(); + + Runnable messageGenerator = () -> { + int counter = 0; + while (generateMessage.get()) { + if (startGenerateMessage.get()) { + counter += 1; + byte[] currentTime = String.valueOf(System.currentTimeMillis()).getBytes(); + messageScheduler.redisScheduleTriggerHandler.messageListener.onMessage( + new DefaultMessage(channelName.getBytes(), currentTime), null); + } + // each message enqueue would lead to one event, so ~250 QPS + TimeoutUtils.sleep(4L); + } + System.out.println("Exiting sent " + counter + " messages "); + }; + scheduler.submit(messageGenerator); + AtomicInteger counter = new AtomicInteger(0); + doAnswer(invocation -> { + counter.incrementAndGet(); + sleep(5); + return System.currentTimeMillis() - Constants.DEFAULT_SCRIPT_EXECUTION_TIME; + }).when(redisTemplate).execute(any(RedisCallback.class)); + + try (MockedStatic threadUtils = Mockito.mockStatic(ThreadUtils.class)) { + threadUtils.when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) + .thenReturn(scheduler); + messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); + waitFor(() -> scheduler.submittedTasks() >= 2, "one start task to be submitted"); + startGenerateMessage.set(true); + // sleep for 2 seconds -> this should send around 500 events + TimeoutUtils.sleep(totalTime); + generateMessage.set(false); // disable Redis pub/sub event + TimeoutUtils.sleep(100); + int expectedMessageMoveCalls = (int) ((totalTime / minDelay) * (1 - buffer)); + messageScheduler.destroy(); + int ranJobs = counter.get(); + int jobCount = scheduler.submittedTasks(); + log.info("Expected Job={}, Ran Jobs={}, Submitted Jobs={}", + expectedMessageMoveCalls, ranJobs, jobCount); + assertTrue(jobCount >= ranJobs); + assertTrue(jobCount >= expectedMessageMoveCalls); + } + } +} diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/RedisScheduleTriggerHandlerTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/RedisScheduleTriggerHandlerTest.java new file mode 100644 index 000000000..4c4eaa4ad --- /dev/null +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/RedisScheduleTriggerHandlerTest.java @@ -0,0 +1,151 @@ +package com.github.sonus21.rqueue.core; + +import com.github.sonus21.TestBase; +import com.github.sonus21.rqueue.CoreUnitTest; +import com.github.sonus21.rqueue.config.RqueueSchedulerConfig; +import com.github.sonus21.rqueue.listener.QueueDetail; +import com.github.sonus21.rqueue.utils.TestUtils; +import com.github.sonus21.rqueue.utils.TimeoutUtils; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.data.redis.connection.DefaultMessage; +import org.springframework.data.redis.connection.MessageListener; + +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +@CoreUnitTest +@Slf4j +class RedisScheduleTriggerHandlerTest extends TestBase { + + private final ExecutorService executor = Executors.newFixedThreadPool(1); + private final String slowQueue = "slow-queue"; + private final QueueDetail slowQueueDetail = TestUtils.createQueueDetail(slowQueue); + private Scheduler scheduler; + @Mock + private RqueueSchedulerConfig rqueueSchedulerConfig; + @Mock + private RqueueRedisListenerContainerFactory rqueueRedisListenerContainerFactory; + private RedisScheduleTriggerHandler redisScheduleTriggerHandler; + private long runningTime = 0; + + + class Task implements Callable { + + @Override + public Void call() throws Exception { + log.info("Running"); + TimeoutUtils.sleep(runningTime); + return null; + } + } + + class Scheduler implements Function> { + + AtomicInteger counter = new AtomicInteger(0); + + @Override + public Future apply(String s) { + counter.incrementAndGet(); + return executor.submit(new Task()); + } + } + + @BeforeEach + public void init() { + MockitoAnnotations.openMocks(this); + EndpointRegistry.delete(); + EndpointRegistry.register(slowQueueDetail); + scheduler = new Scheduler(); + redisScheduleTriggerHandler = new RedisScheduleTriggerHandler(log, + rqueueRedisListenerContainerFactory, rqueueSchedulerConfig, + List.of(slowQueue), scheduler, + (e) -> { + return slowQueueDetail.getScheduledQueueChannelName(); + }); + redisScheduleTriggerHandler.initialize(); + redisScheduleTriggerHandler.startQueue(slowQueue); + } + + @Test + void onMessageListenerTest() throws Exception { + MessageListener messageListener = redisScheduleTriggerHandler.messageListener; + // invalid channel + messageListener.onMessage(new DefaultMessage(slowQueue.getBytes(), "312".getBytes()), null); + TimeoutUtils.sleep(50); + assertEquals(0, scheduler.counter.get()); + + // invalid body + messageListener.onMessage( + new DefaultMessage(slowQueueDetail.getScheduledQueueChannelName().getBytes(), + "sss".getBytes()), null); + TimeoutUtils.sleep(50); + assertEquals(0, scheduler.counter.get()); + + // future time + messageListener.onMessage( + new DefaultMessage(slowQueueDetail.getScheduledQueueChannelName().getBytes(), + String.valueOf(System.currentTimeMillis() + 500).getBytes()), null); + TimeoutUtils.sleep(50); + assertEquals(0, scheduler.counter.get()); + + // both are correct + messageListener.onMessage( + new DefaultMessage(slowQueueDetail.getScheduledQueueChannelName().getBytes(), + String.valueOf(System.currentTimeMillis()).getBytes()), null); + assertEquals(1, scheduler.counter.get()); + // let it run + TimeoutUtils.sleep(100); + + // send another message while one is running + runningTime = 200; + messageListener.onMessage( + new DefaultMessage(slowQueueDetail.getScheduledQueueChannelName().getBytes(), + String.valueOf(System.currentTimeMillis()).getBytes()), null); + assertEquals(2, scheduler.counter.get()); + TimeoutUtils.sleep(100); + + // this should be rejected as another task is already running + messageListener.onMessage( + new DefaultMessage(slowQueueDetail.getScheduledQueueChannelName().getBytes(), + String.valueOf(System.currentTimeMillis()).getBytes()), null); + TimeoutUtils.sleep(50); + assertEquals(2, scheduler.counter.get()); + TimeoutUtils.sleep(100); + + // this should success + long lastRunTime = System.currentTimeMillis(); + messageListener.onMessage( + new DefaultMessage(slowQueueDetail.getScheduledQueueChannelName().getBytes(), + String.valueOf(System.currentTimeMillis()).getBytes()), null); + assertEquals(3, scheduler.counter.get()); + verify(rqueueRedisListenerContainerFactory, times(1)).addMessageListener(any(), any()); + doReturn(400L).when(rqueueSchedulerConfig).getTerminationWaitTime(); + + assertEquals(1, redisScheduleTriggerHandler.queueNameToFuture.size()); + assertEquals(1, redisScheduleTriggerHandler.channelNameToQueueName.size()); + assertTrue(redisScheduleTriggerHandler.queueNameToLastRunTime.get(slowQueue) >= lastRunTime); + + redisScheduleTriggerHandler.stop(); + + assertTrue(redisScheduleTriggerHandler.queueNameToFuture.isEmpty()); + assertEquals(1, redisScheduleTriggerHandler.channelNameToQueueName.size()); + assertEquals(0L, redisScheduleTriggerHandler.queueNameToLastRunTime.get(slowQueue)); + verify(rqueueRedisListenerContainerFactory, times(1)).removeMessageListener(any(), any()); + } +} diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/ScheduledQueueMessageSchedulerTest.java b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/ScheduledQueueMessageSchedulerTest.java index 1a7da35fa..25e91f1ed 100644 --- a/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/ScheduledQueueMessageSchedulerTest.java +++ b/rqueue-core/src/test/java/com/github/sonus21/rqueue/core/ScheduledQueueMessageSchedulerTest.java @@ -36,13 +36,14 @@ import com.github.sonus21.rqueue.config.RqueueSchedulerConfig; import com.github.sonus21.rqueue.listener.QueueDetail; import com.github.sonus21.rqueue.models.event.RqueueBootstrapEvent; +import com.github.sonus21.rqueue.utils.Constants; import com.github.sonus21.rqueue.utils.TestUtils; import com.github.sonus21.rqueue.utils.ThreadUtils; import com.github.sonus21.rqueue.utils.TimeoutUtils; import com.github.sonus21.test.TestTaskScheduler; -import java.util.List; import java.util.Map; -import java.util.Vector; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.lang3.reflect.FieldUtils; import org.junit.jupiter.api.BeforeEach; @@ -56,8 +57,6 @@ import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.TooManyClusterRedirectionsException; -import org.springframework.data.redis.connection.DefaultMessage; -import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.listener.ChannelTopic; @@ -66,8 +65,6 @@ @SuppressWarnings("unchecked") class ScheduledQueueMessageSchedulerTest extends TestBase { - @InjectMocks - private final TestMessageScheduler messageScheduler = new TestMessageScheduler(); private final String slowQueue = "slow-queue"; private final String fastQueue = "fast-queue"; private final QueueDetail slowQueueDetail = TestUtils.createQueueDetail(slowQueue); @@ -80,6 +77,8 @@ class ScheduledQueueMessageSchedulerTest extends TestBase { private RedisTemplate redisTemplate; @Mock private RqueueRedisListenerContainerFactory rqueueRedisListenerContainerFactory; + @InjectMocks + private TestScheduledQueueMessageScheduler messageScheduler; @BeforeEach public void init() { @@ -91,8 +90,8 @@ public void init() { @Test void getChannelName() { - assertEquals( - slowQueueDetail.getScheduledQueueChannelName(), messageScheduler.getChannelName(slowQueue)); + assertEquals(slowQueueDetail.getScheduledQueueChannelName(), + messageScheduler.getChannelName(slowQueue)); } @Test @@ -104,11 +103,9 @@ void getZsetName() { void getNextScheduleTime() { long currentTime = System.currentTimeMillis(); doReturn(5000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli(); - assertThat( - messageScheduler.getNextScheduleTime(slowQueue, null), + assertThat(messageScheduler.getNextScheduleTime(slowQueue, currentTime, null), greaterThanOrEqualTo(currentTime + 5000L)); - assertThat( - messageScheduler.getNextScheduleTime(fastQueue, currentTime + 1000L), + assertThat(messageScheduler.getNextScheduleTime(fastQueue, currentTime, currentTime + 1000L), greaterThanOrEqualTo(currentTime + 5000L)); } @@ -120,9 +117,8 @@ void afterPropertiesSetWithEmptyQueSet() throws Exception { assertNull(FieldUtils.readField(messageScheduler, "scheduler", true)); assertNull(FieldUtils.readField(messageScheduler, "queueRunningState", true)); assertNull(FieldUtils.readField(messageScheduler, "queueNameToScheduledTask", true)); - assertNull(FieldUtils.readField(messageScheduler, "channelNameToQueueName", true)); - assertNull(FieldUtils.readField(messageScheduler, "queueNameToLastMessageScheduleTime", true)); - assertNull(FieldUtils.readField(messageScheduler, "queueSchedulers", true)); + assertNull(FieldUtils.readField(messageScheduler, "queueNameToNextRunTime", true)); + assertNull(FieldUtils.readField(messageScheduler, "redisScheduleTriggerHandler", true)); } @Test @@ -133,15 +129,12 @@ void start() throws Exception { doReturn(true).when(rqueueSchedulerConfig).isEnabled(); doReturn(1000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli(); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); - Map queueRunningState = - (Map) FieldUtils.readField(messageScheduler, "queueRunningState", true); + Map queueRunningState = (Map) FieldUtils.readField( + messageScheduler, "queueRunningState", true); assertEquals(2, queueRunningState.size()); assertTrue(queueRunningState.get(slowQueue)); - assertEquals( - 2, ((Map) FieldUtils.readField(messageScheduler, "queueNameToScheduledTask", true)).size()); - assertEquals( - 2, ((Map) FieldUtils.readField(messageScheduler, "channelNameToQueueName", true)).size()); - assertEquals(2, ((Map) FieldUtils.readField(messageScheduler, "queueSchedulers", true)).size()); + assertEquals(2, + ((Map) FieldUtils.readField(messageScheduler, "queueNameToScheduledTask", true)).size()); TimeoutUtils.sleep(500L); messageScheduler.destroy(); } @@ -153,14 +146,10 @@ void startAddsChannelToMessageListener() throws Exception { doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); doReturn(true).when(rqueueSchedulerConfig).isEnabled(); doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); - doNothing() - .when(rqueueRedisListenerContainerFactory) - .addMessageListener( - any(), eq(new ChannelTopic(slowQueueDetail.getScheduledQueueChannelName()))); - doNothing() - .when(rqueueRedisListenerContainerFactory) - .addMessageListener( - any(), eq(new ChannelTopic(fastQueueDetail.getScheduledQueueChannelName()))); + doNothing().when(rqueueRedisListenerContainerFactory).addMessageListener(any(), + eq(new ChannelTopic(slowQueueDetail.getScheduledQueueChannelName()))); + doNothing().when(rqueueRedisListenerContainerFactory).addMessageListener(any(), + eq(new ChannelTopic(fastQueueDetail.getScheduledQueueChannelName()))); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); TimeoutUtils.sleep(500L); messageScheduler.destroy(); @@ -176,12 +165,12 @@ void stop() throws Exception { messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); TimeoutUtils.sleep(500L); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", false)); - Map queueRunningState = - (Map) FieldUtils.readField(messageScheduler, "queueRunningState", true); + Map queueRunningState = (Map) FieldUtils.readField( + messageScheduler, "queueRunningState", true); assertEquals(2, queueRunningState.size()); assertFalse(queueRunningState.get(slowQueue)); - assertEquals( - 2, ((Map) FieldUtils.readField(messageScheduler, "channelNameToQueueName", true)).size()); + assertEquals(2, + ((Map) FieldUtils.readField(messageScheduler, "queueNameToNextRunTime", true)).size()); assertTrue( ((Map) FieldUtils.readField(messageScheduler, "queueNameToScheduledTask", true)).isEmpty()); messageScheduler.destroy(); @@ -196,19 +185,17 @@ void destroy() throws Exception { doReturn(1000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli(); TestTaskScheduler scheduler = new TestTaskScheduler(); try (MockedStatic threadUtils = Mockito.mockStatic(ThreadUtils.class)) { - threadUtils - .when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) + threadUtils.when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) .thenReturn(scheduler); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); TimeoutUtils.sleep(500L); messageScheduler.destroy(); - Map queueRunningState = - (Map) FieldUtils.readField(messageScheduler, "queueRunningState", true); + Map queueRunningState = (Map) FieldUtils.readField( + messageScheduler, "queueRunningState", true); assertEquals(2, queueRunningState.size()); assertFalse(queueRunningState.get(slowQueue)); - assertTrue( - ((Map) FieldUtils.readField(messageScheduler, "queueNameToScheduledTask", true)) - .isEmpty()); + assertTrue(((Map) FieldUtils.readField(messageScheduler, "queueNameToScheduledTask", + true)).isEmpty()); assertTrue(scheduler.shutdown); } } @@ -221,8 +208,7 @@ void startSubmitsTask() throws Exception { doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); TestTaskScheduler scheduler = new TestTaskScheduler(); try (MockedStatic threadUtils = Mockito.mockStatic(ThreadUtils.class)) { - threadUtils - .when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) + threadUtils.when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) .thenReturn(scheduler); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); assertTrue(scheduler.submittedTasks() >= 1); @@ -238,13 +224,10 @@ void startSubmitsTaskAndThatGetsExecuted() throws Exception { doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); doReturn(1000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli(); AtomicInteger counter = new AtomicInteger(0); - doAnswer( - invocation -> { - counter.incrementAndGet(); - return null; - }) - .when(redisTemplate) - .execute(any(RedisCallback.class)); + doAnswer(invocation -> { + counter.incrementAndGet(); + return null; + }).when(redisTemplate).execute(any(RedisCallback.class)); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); waitFor(() -> counter.get() >= 1, "scripts are getting executed"); messageScheduler.destroy(); @@ -259,16 +242,12 @@ void onCompletionOfExistingTaskNewTaskShouldBeSubmitted() throws Exception { doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); doReturn(1000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli(); AtomicInteger counter = new AtomicInteger(0); - doAnswer( - invocation -> { - counter.incrementAndGet(); - return null; - }) - .when(redisTemplate) - .execute(any(RedisCallback.class)); + doAnswer(invocation -> { + counter.incrementAndGet(); + return null; + }).when(redisTemplate).execute(any(RedisCallback.class)); TestTaskScheduler scheduler = new TestTaskScheduler(); - threadUtils - .when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) + threadUtils.when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) .thenReturn(scheduler); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); waitFor(() -> counter.get() >= 1, "scripts are getting executed"); @@ -287,17 +266,13 @@ void taskShouldBeScheduledOnFailure() throws Exception { doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); doReturn(10000L).when(rqueueSchedulerConfig).getMaxMessageMoverDelay(); AtomicInteger counter = new AtomicInteger(0); - doAnswer( - invocation -> { - counter.incrementAndGet(); - throw new RedisSystemException("Something is not correct", - new NullPointerException("oops!")); - }) - .when(redisTemplate) - .execute(any(RedisCallback.class)); + doAnswer(invocation -> { + counter.incrementAndGet(); + throw new RedisSystemException("Something is not correct", + new NullPointerException("oops!")); + }).when(redisTemplate).execute(any(RedisCallback.class)); TestTaskScheduler scheduler = new TestTaskScheduler(); - threadUtils - .when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) + threadUtils.when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) .thenReturn(scheduler); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); waitFor(() -> counter.get() >= 1, "scripts are getting executed"); @@ -316,81 +291,49 @@ void continuousTaskFailTask() throws Exception { doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); doReturn(100L).when(rqueueSchedulerConfig).getMaxMessageMoverDelay(); AtomicInteger counter = new AtomicInteger(0); - doAnswer( - invocation -> { - int count = counter.incrementAndGet(); - if (count % 3 == 0) { - throw new RedisSystemException("Something is not correct", - new NullPointerException("oops!")); - } - if (count % 3 == 1) { - throw new RedisConnectionFailureException("Unknown host"); - } - throw new ClusterRedirectException(3, "localhost", 9004, - new TooManyClusterRedirectionsException("too many redirects")); - }) - .when(redisTemplate) - .execute(any(RedisCallback.class)); + doAnswer(invocation -> { + int count = counter.incrementAndGet(); + if (count % 3 == 0) { + throw new RedisSystemException("Something is not correct", + new NullPointerException("oops!")); + } + if (count % 3 == 1) { + throw new RedisConnectionFailureException("Unknown host"); + } + throw new ClusterRedirectException(3, "localhost", 9004, + new TooManyClusterRedirectionsException("too many redirects")); + }).when(redisTemplate).execute(any(RedisCallback.class)); TestTaskScheduler scheduler = new TestTaskScheduler(); - threadUtils - .when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) + threadUtils.when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60)) .thenReturn(scheduler); messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); waitFor(() -> counter.get() >= 10, "scripts are getting executed"); sleep(10); messageScheduler.destroy(); - assertTrue(scheduler.submittedTasks() >= 11); } } - @Test - void onMessageListenerTest() throws Exception { - doReturn(true).when(rqueueSchedulerConfig).isEnabled(); - doReturn(5000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli(); - doReturn(1).when(rqueueSchedulerConfig).getScheduledMessageThreadPoolSize(); - doReturn(true).when(rqueueSchedulerConfig).isAutoStart(); - doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled(); - messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true)); - - MessageListener messageListener = - (MessageListener) FieldUtils.readField(messageScheduler, "messageSchedulerListener", true); - // invalid channel - messageListener.onMessage(new DefaultMessage(slowQueue.getBytes(), "312".getBytes()), null); - TimeoutUtils.sleep(50); - assertEquals(2, messageScheduler.scheduleList.stream().filter(e -> !e).count()); - - // invalid body - messageListener.onMessage( - new DefaultMessage( - slowQueueDetail.getScheduledQueueChannelName().getBytes(), "sss".getBytes()), - null); - TimeoutUtils.sleep(50); - assertEquals(2, messageScheduler.scheduleList.stream().filter(e -> !e).count()); - - TimeoutUtils.sleep(110); - // both are correct - messageListener.onMessage( - new DefaultMessage( - slowQueueDetail.getScheduledQueueChannelName().getBytes(), - String.valueOf(System.currentTimeMillis()).getBytes()), - null); - assertEquals(3, messageScheduler.scheduleList.stream().filter(e -> !e).count()); - messageScheduler.destroy(); - } + static class TestScheduledQueueMessageScheduler extends ScheduledQueueMessageScheduler { - static class TestMessageScheduler extends ScheduledQueueMessageScheduler { + final AtomicInteger scheduleCounter; + final AtomicInteger addTaskCounter; - List scheduleList; + TestScheduledQueueMessageScheduler() { + this.scheduleCounter = new AtomicInteger(0); + this.addTaskCounter = new AtomicInteger(0); + } - TestMessageScheduler() { - this.scheduleList = new Vector<>(); + @Override + protected void schedule(String queueName) { + scheduleCounter.incrementAndGet(); + super.schedule(queueName); } @Override - protected synchronized void schedule(String queueName, Long startTime, boolean forceSchedule) { - super.schedule(queueName, startTime, forceSchedule); - this.scheduleList.add(forceSchedule); + protected Future addTask(String queueName) { + addTaskCounter.incrementAndGet(); + return super.addTask(queueName); } } } diff --git a/rqueue-spring-boot-example/src/main/java/com/github/sonus21/rqueue/example/MessageListener.java b/rqueue-spring-boot-example/src/main/java/com/github/sonus21/rqueue/example/MessageListener.java index acedfcacf..467b6f112 100644 --- a/rqueue-spring-boot-example/src/main/java/com/github/sonus21/rqueue/example/MessageListener.java +++ b/rqueue-spring-boot-example/src/main/java/com/github/sonus21/rqueue/example/MessageListener.java @@ -65,31 +65,19 @@ public void onSimpleMessage(String message) { execute("simple: {}", message, false); } - @RqueueListener( - value = {"${rqueue.delay.queue}", "${rqueue.delay2.queue}"}, - numRetries = "${rqueue.delay.queue.retries}", - visibilityTimeout = "60*60*1000") + @RqueueListener(value = {"${rqueue.delay.queue}", + "${rqueue.delay2.queue}"}, numRetries = "${rqueue.delay.queue.retries}", visibilityTimeout = "60*60*1000") public void onMessage(String message) { execute("delay: {}", message, true); } - @RqueueListener( - value = "job-queue", - deadLetterQueue = "job-morgue", - numRetries = "2", - deadLetterQueueListenerEnabled = "false", - concurrency = "10-20") + @RqueueListener(value = "job-queue", deadLetterQueue = "job-morgue", numRetries = "2", deadLetterQueueListenerEnabled = "false", concurrency = "10-20") public void onJobMessage(Job job) { execute("job-queue: {}", job, true); } - @RqueueListener( - value = "sch-job-queue", - deadLetterQueue = "job-morgue", - numRetries = "2", - deadLetterQueueListenerEnabled = "false", - concurrency = "1-3") + @RqueueListener(value = "sch-job-queue", deadLetterQueue = "job-morgue", numRetries = "2", deadLetterQueueListenerEnabled = "false", concurrency = "1-3") public void onSchJobMessage(Job job, @Header(RqueueMessageHeaders.ID) String messageId) { execute("sch-job-queue: {}", job, false); count += 1; diff --git a/rqueue-spring-boot-example/src/main/resources/logback.xml b/rqueue-spring-boot-example/src/main/resources/logback.xml index e77d8d73b..d6a612743 100644 --- a/rqueue-spring-boot-example/src/main/resources/logback.xml +++ b/rqueue-spring-boot-example/src/main/resources/logback.xml @@ -17,16 +17,12 @@ - [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] %-5level %X{traceId:-} %X{spanId:-} - ${PID:-} %logger{36} - %msg%n - + [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] %-5level %X{traceId:-} %X{spanId:-} ${PID:-} %logger{36} - %msg%n - [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] %-5level %X{traceId:-} %X{spanId:-} - ${PID:-} %logger{36} - %msg%n - + [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] %-5level %X{traceId:-} %X{spanId:-} ${PID:-} %logger{36} - %msg%n log/app.log @@ -39,4 +35,10 @@ + + + + + + \ No newline at end of file diff --git a/rqueue-test-util/src/main/java/com/github/sonus21/test/TestTaskScheduler.java b/rqueue-test-util/src/main/java/com/github/sonus21/test/TestTaskScheduler.java index 7572bbdb2..106357b68 100644 --- a/rqueue-test-util/src/main/java/com/github/sonus21/test/TestTaskScheduler.java +++ b/rqueue-test-util/src/main/java/com/github/sonus21/test/TestTaskScheduler.java @@ -16,7 +16,7 @@ package com.github.sonus21.test; -import java.time.Instant; +import java.time.Duration; import java.util.List; import java.util.Vector; import java.util.concurrent.Future; @@ -25,7 +25,6 @@ public class TestTaskScheduler extends ThreadPoolTaskScheduler { - private static final long serialVersionUID = 3617860362304703358L; public boolean shutdown = false; List> tasks = new Vector<>(); @@ -34,6 +33,11 @@ public TestTaskScheduler() { afterPropertiesSet(); } + public TestTaskScheduler(int poolSize) { + setPoolSize(poolSize); + afterPropertiesSet(); + } + @Override public Future submit(Runnable r) { Future f = super.submit(r); @@ -42,8 +46,8 @@ public Future submit(Runnable r) { } @Override - public ScheduledFuture schedule(Runnable r, Instant instant) { - ScheduledFuture f = super.schedule(r, instant); + public ScheduledFuture scheduleAtFixedRate(Runnable task, Duration period) { + ScheduledFuture f = super.scheduleAtFixedRate(task, period); tasks.add(f); return f; } diff --git a/rqueue-test-util/src/main/resources/logback.xml b/rqueue-test-util/src/main/resources/logback.xml index ff31df820..9d0ffeb50 100644 --- a/rqueue-test-util/src/main/resources/logback.xml +++ b/rqueue-test-util/src/main/resources/logback.xml @@ -17,16 +17,12 @@ - %property{testClass}::%property{testName} [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] - %-5level %logger{36} - %msg%n - + %property{testClass}::%property{testName} [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] %-5level %logger{36} - %msg%n - %property{testClass}::%property{testName} [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] - %-5level %logger{36} - %msg%n - + %property{testClass}::%property{testName} [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] %-5level %logger{36} - %msg%n log/test.log @@ -38,9 +34,7 @@ - %property{testClass}::%property{testName} [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] - %-5level %logger{36} - %msg%n - + %property{testClass}::%property{testName} [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] %-5level %logger{36} - %msg%n log/monitor.log @@ -52,9 +46,7 @@ - %property{testClass}::%property{testName} [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] - %-5level %logger{36} - %msg%n - + %property{testClass}::%property{testName} [%date{dd-MM-yyyy HH:mm:ss.SSS}] [%thread] %-5level %logger{36} - %msg%n log/message.log @@ -99,5 +91,11 @@ + + + + + + \ No newline at end of file