From 2d50f7de157e8bff6e195753e1e559fb9f45fc85 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Mon, 31 Aug 2026 14:56:41 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=AE=8C=E5=96=84=E5=88=86=E5=B8=83?= =?UTF-8?q?=E5=BC=8F=E8=B0=83=E5=BA=A6=E6=81=A2=E5=A4=8D=E4=B8=8E=E6=89=B9?= =?UTF-8?q?=E9=87=8F=E8=A7=A6=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 持久化 Quartz refire 状态并收口运行时启动关闭顺序 - 增加批量 Trigger 获取配置、校验与回归测试 --- .../scheduler/ScheduleRefireException.java | 53 +++++ .../scheduler/quartz/AbstractDispatchJob.java | 3 + .../scheduler/quartz/QuartzRuntime.java | 121 +++++++++++- .../quartz/QuartzScheduleMapper.java | 30 +++ .../quartz/QuartzScheduleService.java | 3 +- .../quartz/QuartzSchedulerConfig.java | 18 ++ .../quartz/QuartzSchedulerFactory.java | 14 +- .../quartz/QuartzJdbcIntegrationTest.java | 82 ++++++++ .../scheduler/quartz/QuartzRuntimeTest.java | 105 ++++++++++ .../quartz/QuartzScheduleServiceTest.java | 182 ++++++++++++++++-- .../quartz/QuartzSchedulerFactoryTest.java | 40 ++++ .../pom.xml | 5 + .../EasyAgentsSchedulerAutoConfiguration.java | 4 + .../boot/EasyAgentsSchedulerProperties.java | 45 +++++ ...yAgentsSchedulerAutoConfigurationTest.java | 48 +++++ 15 files changed, 735 insertions(+), 18 deletions(-) create mode 100644 easy-agents-scheduler/easy-agents-scheduler-core/src/main/java/com/easyagents/scheduler/ScheduleRefireException.java create mode 100644 easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzRuntimeTest.java create mode 100644 easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactoryTest.java diff --git a/easy-agents-scheduler/easy-agents-scheduler-core/src/main/java/com/easyagents/scheduler/ScheduleRefireException.java b/easy-agents-scheduler/easy-agents-scheduler-core/src/main/java/com/easyagents/scheduler/ScheduleRefireException.java new file mode 100644 index 0000000..01ef029 --- /dev/null +++ b/easy-agents-scheduler/easy-agents-scheduler-core/src/main/java/com/easyagents/scheduler/ScheduleRefireException.java @@ -0,0 +1,53 @@ +package com.easyagents.scheduler; + +import java.time.Duration; + +/** + * 请求调度提供方持久化重试同一次逻辑触发的异常。 + * + *

仅用于尚未产生业务副作用、且调用方必须保证至少一次登记的短 Handler。 + * {@code maxRefires=0} 表示不限制持久化重试次数；Provider 正常情况下应释放当前执行线程， + * 并通过持久化的延迟触发保留原 fire 上下文。仅当 JobStore 无法写入持久重试时， + * Provider 可按相同退避暂时占用当前线程并原地重试，以避免正常完成造成 fire 丢失。

+ */ +public class ScheduleRefireException extends RuntimeException { + + private static final int DEFAULT_MAX_REFIRES = 0; + private static final Duration DEFAULT_BASE_DELAY = Duration.ofMillis(250); + private final int maxRefires; + private final Duration baseDelay; + + public ScheduleRefireException(String message, Throwable cause) { + this(message, cause, DEFAULT_MAX_REFIRES, DEFAULT_BASE_DELAY); + } + + public ScheduleRefireException(String message, Throwable cause, + int maxRefires, Duration baseDelay) { + super(requireMessage(message), cause); + if (maxRefires < 0) throw new IllegalArgumentException("maxRefires must not be negative"); + if (baseDelay == null || baseDelay.isZero() || baseDelay.isNegative()) { + throw new IllegalArgumentException("baseDelay must be positive"); + } + this.maxRefires = maxRefires; + this.baseDelay = baseDelay; + } + + /** + * @return 最大重新执行次数，0 表示不限制 + */ + public int maxRefires() { + return maxRefires; + } + + public Duration delayFor(int refireNumber) { + int shift = Math.min(4, Math.max(0, refireNumber - 1)); + return baseDelay.multipliedBy(1L << shift); + } + + private static String requireMessage(String message) { + if (message == null || message.isBlank()) { + throw new IllegalArgumentException("message must not be blank"); + } + return message; + } +} diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java index 2f73903..22e8c00 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/AbstractDispatchJob.java @@ -29,6 +29,9 @@ abstract class AbstractDispatchJob implements InterruptableJob { throw new JobExecutionException("easy-agents scheduler runtime is not available"); } quartzRuntime.execute(context); + } catch (JobExecutionException exception) { + // JobExecutionException 继承 SchedulerException，必须保留 refire 等控制语义。 + throw exception; } catch (SchedulerException exception) { throw new JobExecutionException("failed to access scheduler runtime", exception, false); } finally { diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java index 2d5b30d..c7e3081 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzRuntime.java @@ -7,9 +7,16 @@ import com.easyagents.scheduler.ScheduleExecutionListener; import com.easyagents.scheduler.ScheduleFireContext; import com.easyagents.scheduler.ScheduleHandler; import com.easyagents.scheduler.ScheduleId; +import com.easyagents.scheduler.ScheduleRefireException; import org.quartz.JobDataMap; import org.quartz.JobExecutionContext; import org.quartz.JobExecutionException; +import org.quartz.ObjectAlreadyExistsException; +import org.quartz.SchedulerException; +import org.quartz.SimpleScheduleBuilder; +import org.quartz.Trigger; +import org.quartz.TriggerBuilder; +import org.quartz.TriggerKey; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -19,6 +26,10 @@ import java.util.Collection; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.nio.charset.StandardCharsets; +import java.util.Date; +import java.util.UUID; +import java.util.concurrent.locks.LockSupport; /** * 当前 Scheduler 节点的 Handler 和监听器运行时。 @@ -55,15 +66,27 @@ final class QuartzRuntime { void execute(JobExecutionContext quartzContext) throws JobExecutionException { JobDataMap data = quartzContext.getMergedJobDataMap(); ScheduleDefinition definition = QuartzScheduleMapper.toDefinition(data); - Instant actualFireTime = toInstant(quartzContext.getFireTime(), Instant.now()); + Instant observedActualFireTime = toInstant(quartzContext.getFireTime(), Instant.now()); + Instant scheduledFireTime = instantValue( + data, + QuartzScheduleMapper.KEY_RETRY_SCHEDULED_FIRE_TIME, + toInstant(quartzContext.getScheduledFireTime(), observedActualFireTime) + ); + Instant actualFireTime = instantValue( + data, + QuartzScheduleMapper.KEY_RETRY_ACTUAL_FIRE_TIME, + observedActualFireTime + ); ScheduleFireContext context = new ScheduleFireContext( definition.id(), definition.handlerCode(), - toInstant(quartzContext.getScheduledFireTime(), actualFireTime), + scheduledFireTime, actualFireTime, - quartzContext.getFireInstanceId(), + stringValue(data, QuartzScheduleMapper.KEY_RETRY_FIRE_INSTANCE_ID, + quartzContext.getFireInstanceId()), stringValue(data, QuartzScheduleMapper.KEY_INVOCATION), - quartzContext.isRecovering(), + booleanValue(data, QuartzScheduleMapper.KEY_RETRY_RECOVERING, + quartzContext.isRecovering()), definition.parameters() ); ScheduleHandler handler = handlers.get(definition.handlerCode()); @@ -82,10 +105,80 @@ final class QuartzRuntime { notifySucceeded(context, elapsed(startedAt)); } catch (Exception exception) { notifyFailed(context, elapsed(startedAt), exception); + if (exception instanceof ScheduleRefireException retry) { + int attempt = Math.max( + intValue(data, QuartzScheduleMapper.KEY_RETRY_ATTEMPT, 0), + quartzContext.getRefireCount() + ); + if (retry.maxRefires() == 0 || attempt < retry.maxRefires()) { + try { + persistRetry(quartzContext, context, retry, attempt + 1); + return; + } catch (SchedulerException retryFailure) { + exception.addSuppressed(retryFailure); + log.error("Failed to persist schedule handler retry: scheduleId={}, attempt={}", + context.scheduleId(), attempt + 1, retryFailure); + if (canRefireInPlace(quartzContext, retry, attempt + 1)) { + // JobStore 暂时不可写时，正常完成原 fire 会造成登记丢失。 + // 仅在这条降级路径短时占用当前 worker；一旦持久重试落库即释放。 + throw new JobExecutionException(exception, true); + } + } + } else { + log.error("Schedule handler retry limit exhausted: scheduleId={}, retries={}", + context.scheduleId(), retry.maxRefires(), exception); + } + } + // 不在 Quartz worker 内无限 refire。持久化重试失败时保留原异常， + // 由 requestRecovery 和集群故障恢复处理未完成的 fired trigger。 throw new JobExecutionException(exception, false); } } + private static boolean canRefireInPlace( + JobExecutionContext context, + ScheduleRefireException retry, + int attempt + ) { + LockSupport.parkNanos(retry.delayFor(attempt).toNanos()); + if (Thread.currentThread().isInterrupted()) return false; + try { + return !context.getScheduler().isShutdown(); + } catch (SchedulerException exception) { + return false; + } + } + + private static void persistRetry( + JobExecutionContext quartzContext, + ScheduleFireContext context, + ScheduleRefireException retry, + int attempt + ) throws SchedulerException { + String source = context.scheduleId() + "|" + context.fireInstanceId() + "|" + + String.valueOf(context.invocationId()) + "|" + attempt; + String retryName = "retry-" + UUID.nameUUIDFromBytes( + source.getBytes(StandardCharsets.UTF_8)); + TriggerKey retryKey = new TriggerKey( + retryName, + QuartzScheduleMapper.GROUP_PREFIX + "retry." + context.scheduleId().namespace() + ); + Trigger trigger = TriggerBuilder.newTrigger() + .withIdentity(retryKey) + .forJob(quartzContext.getJobDetail().getKey()) + .usingJobData(QuartzScheduleMapper.retryData(context, attempt)) + .startAt(Date.from(Instant.now().plus(retry.delayFor(attempt)))) + .withSchedule(SimpleScheduleBuilder.simpleSchedule() + .withRepeatCount(0) + .withMisfireHandlingInstructionFireNow()) + .build(); + try { + quartzContext.getScheduler().scheduleJob(trigger); + } catch (ObjectAlreadyExistsException ignored) { + // 同一原始 fire/attempt 的确定性 key 已落库，即视为持久化成功。 + } + } + /** * 向监听器发布 Misfire 事件。 * @@ -175,4 +268,24 @@ final class QuartzRuntime { Object value = data.get(key); return value == null ? null : value.toString(); } + + private static String stringValue(JobDataMap data, String key, String fallback) { + String value = stringValue(data, key); + return value == null ? fallback : value; + } + + private static Instant instantValue(JobDataMap data, String key, Instant fallback) { + String value = stringValue(data, key); + return value == null ? fallback : Instant.ofEpochMilli(Long.parseLong(value)); + } + + private static boolean booleanValue(JobDataMap data, String key, boolean fallback) { + String value = stringValue(data, key); + return value == null ? fallback : Boolean.parseBoolean(value); + } + + private static int intValue(JobDataMap data, String key, int fallback) { + String value = stringValue(data, key); + return value == null ? fallback : Integer.parseInt(value); + } } diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java index 9a59b71..f6666ae 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleMapper.java @@ -7,6 +7,7 @@ import com.easyagents.scheduler.OnceSchedulePlan; import com.easyagents.scheduler.ScheduleDefinition; import com.easyagents.scheduler.ScheduleErrorCode; import com.easyagents.scheduler.ScheduleException; +import com.easyagents.scheduler.ScheduleFireContext; import com.easyagents.scheduler.ScheduleId; import com.easyagents.scheduler.SchedulePlan; import org.quartz.CronScheduleBuilder; @@ -47,6 +48,11 @@ final class QuartzScheduleMapper { static final String KEY_DESCRIPTION = "ea.description"; static final String KEY_INVOCATION = "ea.invocationId"; static final String KEY_IMMEDIATE_PARAMETER_SNAPSHOT = "ea.immediateParameterSnapshot"; + static final String KEY_RETRY_ATTEMPT = "ea.retry.attempt"; + static final String KEY_RETRY_SCHEDULED_FIRE_TIME = "ea.retry.scheduledFireTime"; + static final String KEY_RETRY_ACTUAL_FIRE_TIME = "ea.retry.actualFireTime"; + static final String KEY_RETRY_FIRE_INSTANCE_ID = "ea.retry.fireInstanceId"; + static final String KEY_RETRY_RECOVERING = "ea.retry.recovering"; static final String PARAMETER_PREFIX = "ea.parameter."; static final String IMMEDIATE_PARAMETER_PREFIX = "ea.immediateParameter."; static final String GROUP_PREFIX = "ea.scheduler."; @@ -153,6 +159,30 @@ final class QuartzScheduleMapper { return data; } + /** + * 固化持久化重试所需的原始 fire 上下文。 + */ + static JobDataMap retryData(ScheduleFireContext context, int attempt) { + JobDataMap data = identityData(context.scheduleId()); + // Trigger 数据覆盖 JobDetail；任务定义被 replace 后，既有 fire 的重试仍须 + // 派发给首次触发时的 Handler，而不是意外切换到新 Handler。 + data.put(KEY_HANDLER, context.handlerCode()); + data.put(KEY_RETRY_ATTEMPT, Integer.toString(attempt)); + data.put(KEY_RETRY_SCHEDULED_FIRE_TIME, + Long.toString(context.scheduledFireTime().toEpochMilli())); + data.put(KEY_RETRY_ACTUAL_FIRE_TIME, + Long.toString(context.actualFireTime().toEpochMilli())); + data.put(KEY_RETRY_FIRE_INSTANCE_ID, context.fireInstanceId()); + data.put(KEY_RETRY_RECOVERING, Boolean.toString(context.recovering())); + if (context.invocationId() != null) { + data.put(KEY_INVOCATION, context.invocationId()); + } + // 重试必须沿用首次 fire 的参数快照，不能读取期间被替换的新定义参数。 + data.put(KEY_IMMEDIATE_PARAMETER_SNAPSHOT, Boolean.TRUE.toString()); + putParameters(data, context.parameters(), IMMEDIATE_PARAMETER_PREFIX); + return data; + } + /** * 从持久 JobData 恢复公共定义。 * diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java index a47f42c..078dd37 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzScheduleService.java @@ -335,7 +335,8 @@ public final class QuartzScheduleService implements ScheduleService, AutoCloseab } try { if (!scheduler.isShutdown()) { - // false 保证 close 本身有界；Quartz 会向内部 InterruptableJob 发送中断。 + // Factory 启用 interruptJobsOnShutdown，Quartz 会先中断内部 + // InterruptableJob；false 只表示不再无界等待忽略中断的 Handler。 scheduler.shutdown(false); } } catch (SchedulerException exception) { diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java index 070f38f..35c9f48 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerConfig.java @@ -10,6 +10,8 @@ package com.easyagents.scheduler.quartz; * @param clustered 是否启用 JDBC 集群 * @param threadCount Quartz Worker 线程数 * @param threadPriority Quartz Worker 线程优先级 + * @param batchTriggerAcquisitionMaxCount 单次批量获取 Trigger 的最大数量 + * @param batchTriggerAcquisitionFireAheadTimeWindowMillis 可提前纳入批量的时间窗口，单位毫秒 * @param clusterCheckinIntervalMillis 集群心跳间隔，单位毫秒 * @param misfireThresholdMillis Misfire 判定阈值，单位毫秒 * @param waitForJobsToCompleteOnShutdown 关闭时是否等待运行中任务完成 @@ -24,6 +26,8 @@ public record QuartzSchedulerConfig( boolean clustered, int threadCount, int threadPriority, + int batchTriggerAcquisitionMaxCount, + long batchTriggerAcquisitionFireAheadTimeWindowMillis, long clusterCheckinIntervalMillis, long misfireThresholdMillis, boolean waitForJobsToCompleteOnShutdown, @@ -55,6 +59,18 @@ public record QuartzSchedulerConfig( if (threadPriority < Thread.MIN_PRIORITY || threadPriority > Thread.MAX_PRIORITY) { throw new IllegalArgumentException("threadPriority must be between 1 and 10"); } + if (batchTriggerAcquisitionMaxCount < 1 + || batchTriggerAcquisitionMaxCount > threadCount) { + throw new IllegalArgumentException( + "batchTriggerAcquisitionMaxCount must be between 1 and threadCount" + ); + } + if (batchTriggerAcquisitionFireAheadTimeWindowMillis < 0 + || batchTriggerAcquisitionFireAheadTimeWindowMillis > 60_000L) { + throw new IllegalArgumentException( + "batchTriggerAcquisitionFireAheadTimeWindowMillis must be between 0 and 60000" + ); + } if (clusterCheckinIntervalMillis < 1000) { throw new IllegalArgumentException( "clusterCheckinIntervalMillis must be at least 1000" @@ -85,6 +101,8 @@ public record QuartzSchedulerConfig( true, 8, Thread.NORM_PRIORITY, + 1, + 0L, 15_000L, 60_000L, true, diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java index 57c206f..7460ebd 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/main/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactory.java @@ -154,7 +154,7 @@ public final class QuartzSchedulerFactory { * @param dataSourceName Quartz 内部 DataSource 名称 * @return Quartz 属性 */ - private static Properties properties( + static Properties properties( QuartzSchedulerConfig config, String dataSourceName ) { @@ -173,6 +173,14 @@ public final class QuartzSchedulerFactory { "org.quartz.threadPool.threadPriority", Integer.toString(config.threadPriority()) ); + properties.setProperty( + "org.quartz.scheduler.batchTriggerAcquisitionMaxCount", + Integer.toString(config.batchTriggerAcquisitionMaxCount()) + ); + properties.setProperty( + "org.quartz.scheduler.batchTriggerAcquisitionFireAheadTimeWindow", + Long.toString(config.batchTriggerAcquisitionFireAheadTimeWindowMillis()) + ); properties.setProperty( "org.quartz.jobStore.class", "org.quartz.impl.jdbcjobstore.JobStoreTX" @@ -182,6 +190,10 @@ public final class QuartzSchedulerFactory { config.driverDelegateClass() ); properties.setProperty("org.quartz.jobStore.useProperties", "true"); + properties.setProperty( + "org.quartz.jobStore.acquireTriggersWithinLock", + Boolean.toString(config.batchTriggerAcquisitionMaxCount() > 1) + ); properties.setProperty("org.quartz.jobStore.dataSource", dataSourceName); properties.setProperty("org.quartz.jobStore.tablePrefix", config.tablePrefix()); properties.setProperty( diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java index c7d9fac..6b90107 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzJdbcIntegrationTest.java @@ -9,6 +9,7 @@ import com.easyagents.scheduler.ScheduleException; import com.easyagents.scheduler.ScheduleFireContext; import com.easyagents.scheduler.ScheduleHandler; import com.easyagents.scheduler.ScheduleId; +import com.easyagents.scheduler.ScheduleRefireException; import org.h2.jdbcx.JdbcDataSource; import org.h2.tools.RunScript; import org.junit.Test; @@ -20,9 +21,11 @@ import java.sql.Connection; import java.sql.ResultSet; import java.sql.Statement; import java.time.Instant; +import java.time.Duration; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -218,6 +221,83 @@ public class QuartzJdbcIntegrationTest { } } + /** + * 验证延迟重试先落入 JDBC JobStore，关闭并重建节点后仍使用原 fire 上下文执行。 + */ + @Test + public void shouldRecoverPersistedRetryAfterSchedulerRestart() throws Exception { + JdbcDataSource dataSource = dataSource(); + executeSchema(dataSource); + QuartzSchedulerConfig config = jdbcConfig(); + CountDownLatch firstFailed = new CountDownLatch(1); + CountDownLatch retryCompleted = new CountDownLatch(1); + CopyOnWriteArrayList contexts = new CopyOnWriteArrayList<>(); + ScheduleHandler retryingHandler = new ScheduleHandler() { + @Override + public String code() { + return "persistent-retry-handler"; + } + + @Override + public void execute(ScheduleFireContext context) { + contexts.add(context); + if (contexts.size() == 1) { + firstFailed.countDown(); + throw new ScheduleRefireException("temporary database failure", + new IllegalStateException("unavailable"), 2, + Duration.ofSeconds(2)); + } + retryCompleted.countDown(); + } + }; + ScheduleId scheduleId = new ScheduleId("jdbc", "persistent-retry"); + QuartzScheduleService first = QuartzSchedulerFactory.createJdbc( + dataSource, config, List.of(retryingHandler), List.of()); + first.start(); + first.create(new ScheduleDefinition( + scheduleId, + retryingHandler.code(), + new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")), + MisfirePolicy.FIRE_ONCE_NOW, + ConcurrencyPolicy.DISALLOW, + true, + Map.of("snapshot", "original"), + "persistent retry" + )); + first.triggerNow(scheduleId, "persistent-invocation", Map.of()); + assertTrue(firstFailed.await(5, TimeUnit.SECONDS)); + awaitRetryTrigger(dataSource); + first.close(); + + QuartzScheduleService restarted = QuartzSchedulerFactory.createJdbc( + dataSource, config, List.of(retryingHandler), List.of()); + try { + restarted.start(); + assertTrue("persisted retry did not execute after restart", + retryCompleted.await(8, TimeUnit.SECONDS)); + assertEquals(contexts.get(0).scheduledFireTime(), contexts.get(1).scheduledFireTime()); + assertEquals(contexts.get(0).actualFireTime(), contexts.get(1).actualFireTime()); + assertEquals(contexts.get(0).fireInstanceId(), contexts.get(1).fireInstanceId()); + assertEquals("persistent-invocation", contexts.get(1).invocationId()); + } finally { + restarted.close(); + } + } + + private static void awaitRetryTrigger(JdbcDataSource dataSource) throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + do { + try (Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery( + "SELECT COUNT(*) FROM QRTZ_TRIGGERS WHERE TRIGGER_GROUP LIKE 'ea.scheduler.retry.%'")) { + if (result.next() && result.getInt(1) > 0) return; + } + Thread.sleep(25L); + } while (System.nanoTime() < deadline); + fail("persistent retry trigger was not stored"); + } + private static JdbcDataSource dataSource() { JdbcDataSource dataSource = new JdbcDataSource(); dataSource.setURL("jdbc:h2:mem:scheduler-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1"); @@ -259,6 +339,8 @@ public class QuartzJdbcIntegrationTest { false, 2, Thread.NORM_PRIORITY, + 1, + 0L, 15_000L, 1_000L, true, diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzRuntimeTest.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzRuntimeTest.java new file mode 100644 index 0000000..486980e --- /dev/null +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzRuntimeTest.java @@ -0,0 +1,105 @@ +package com.easyagents.scheduler.quartz; + +import com.easyagents.scheduler.ConcurrencyPolicy; +import com.easyagents.scheduler.MisfirePolicy; +import com.easyagents.scheduler.OnceSchedulePlan; +import com.easyagents.scheduler.ScheduleDefinition; +import com.easyagents.scheduler.ScheduleHandler; +import com.easyagents.scheduler.ScheduleId; +import com.easyagents.scheduler.ScheduleRefireException; +import org.junit.Test; +import org.quartz.JobDetail; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; +import org.quartz.Scheduler; +import org.quartz.SchedulerException; +import org.quartz.Trigger; + +import java.lang.reflect.Proxy; +import java.time.Duration; +import java.time.Instant; +import java.util.Date; +import java.util.List; +import java.util.Map; + +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +/** {@link QuartzRuntime} 失败控制语义测试。 */ +public class QuartzRuntimeTest { + + /** JobStore 无法保存延迟触发时不得把当前 fire 当作正常完成。 */ + @Test + public void shouldRefireInPlaceWhenPersistentRetryCannotBeStored() throws Exception { + ScheduleDefinition definition = new ScheduleDefinition( + new ScheduleId("test", "retry-store-failure"), + "handler", + new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")), + MisfirePolicy.FIRE_ONCE_NOW, + ConcurrencyPolicy.DISALLOW, + true, + Map.of(), + "retry store failure" + ); + JobDetail job = QuartzScheduleMapper.toJobDetail(definition); + Scheduler scheduler = proxy(Scheduler.class, (proxy, method, arguments) -> { + if ("scheduleJob".equals(method.getName()) + && arguments != null && arguments.length == 1 + && arguments[0] instanceof Trigger) { + throw new SchedulerException("job store unavailable"); + } + if ("isShutdown".equals(method.getName())) return false; + return defaultValue(method.getReturnType()); + }); + Date fireTime = new Date(); + JobExecutionContext context = proxy(JobExecutionContext.class, + (proxy, method, arguments) -> switch (method.getName()) { + case "getMergedJobDataMap" -> job.getJobDataMap(); + case "getJobDetail" -> job; + case "getScheduler" -> scheduler; + case "getFireTime", "getScheduledFireTime" -> fireTime; + case "getFireInstanceId" -> "fire-1"; + case "isRecovering" -> false; + case "getRefireCount" -> 0; + default -> defaultValue(method.getReturnType()); + }); + QuartzRuntime runtime = new QuartzRuntime(List.of(new ScheduleHandler() { + @Override + public String code() { + return "handler"; + } + + @Override + public void execute(com.easyagents.scheduler.ScheduleFireContext context) { + throw new ScheduleRefireException("registration unavailable", + new IllegalStateException("database unavailable"), 0, + Duration.ofMillis(1)); + } + }), List.of()); + + try { + runtime.execute(context); + fail("expected refire request"); + } catch (JobExecutionException exception) { + assertTrue(exception.refireImmediately()); + } + } + + @SuppressWarnings("unchecked") + private static T proxy(Class type, java.lang.reflect.InvocationHandler handler) { + return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class[]{type}, handler); + } + + private static Object defaultValue(Class type) { + if (!type.isPrimitive()) return null; + if (type == boolean.class) return false; + if (type == byte.class) return (byte) 0; + if (type == short.class) return (short) 0; + if (type == int.class) return 0; + if (type == long.class) return 0L; + if (type == float.class) return 0F; + if (type == double.class) return 0D; + if (type == char.class) return '\0'; + return null; + } +} diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java index 238c0ed..8559542 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzScheduleServiceTest.java @@ -10,20 +10,25 @@ import com.easyagents.scheduler.ScheduleException; import com.easyagents.scheduler.ScheduleFireContext; import com.easyagents.scheduler.ScheduleHandler; import com.easyagents.scheduler.ScheduleId; +import com.easyagents.scheduler.ScheduleRefireException; import com.easyagents.scheduler.ScheduleStatus; import org.junit.After; import org.junit.Test; import org.quartz.JobDetail; import org.quartz.Scheduler; import org.quartz.impl.StdSchedulerFactory; +import org.quartz.impl.matchers.GroupMatcher; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.time.Instant; import java.time.ZoneId; import java.util.Base64; +import java.util.List; import java.util.Map; import java.util.Properties; import java.util.UUID; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -112,6 +117,108 @@ public class QuartzScheduleServiceTest { assertEquals("value", captured.get().parameters().get("stable")); } + /** 验证可恢复失败会持久化延迟重试，并保留原始 fire 上下文。 */ + @Test + public void shouldRefireRetryableHandlerFailure() throws Exception { + AtomicInteger attempts = new AtomicInteger(); + CountDownLatch completed = new CountDownLatch(1); + CopyOnWriteArrayList contexts = new CopyOnWriteArrayList<>(); + service = newRamService(context -> { + contexts.add(context); + if (attempts.incrementAndGet() == 1) { + throw new ScheduleRefireException("temporary registration failure", + new IllegalStateException("database unavailable"), 2, + Duration.ofMillis(100)); + } + completed.countDown(); + }); + ScheduleDefinition definition = onceDefinition( + "retryable-immediate", + MisfirePolicy.FIRE_ONCE_NOW, + ConcurrencyPolicy.DISALLOW, + Instant.parse("2099-01-01T00:00:00Z") + ); + service.create(definition); + + service.triggerNow(definition.id(), "retryable-invocation", Map.of()); + + assertTrue("retryable handler was not refired", completed.await(5, TimeUnit.SECONDS)); + assertEquals(2, attempts.get()); + assertEquals(contexts.get(0).scheduledFireTime(), contexts.get(1).scheduledFireTime()); + assertEquals(contexts.get(0).actualFireTime(), contexts.get(1).actualFireTime()); + assertEquals(contexts.get(0).fireInstanceId(), contexts.get(1).fireInstanceId()); + assertEquals("retryable-invocation", contexts.get(1).invocationId()); + } + + /** 持久化延迟重试不得占用当前 Quartz worker。 */ + @Test + public void shouldReleaseWorkerWhilePersistentRetryIsDelayed() throws Exception { + CountDownLatch retryScheduled = new CountDownLatch(1); + CountDownLatch healthyCompleted = new CountDownLatch(1); + CountDownLatch retryCompleted = new CountDownLatch(1); + AtomicInteger retryAttempts = new AtomicInteger(); + service = newRamService(context -> { + if (context.scheduleId().name().equals("delayed-retry")) { + if (retryAttempts.incrementAndGet() == 1) { + retryScheduled.countDown(); + throw new ScheduleRefireException("temporary registration failure", + new IllegalStateException("database unavailable"), 1, + Duration.ofSeconds(1)); + } + retryCompleted.countDown(); + } else { + healthyCompleted.countDown(); + } + }, 30_000L, 1); + ScheduleDefinition retry = onceDefinition( + "delayed-retry", MisfirePolicy.FIRE_ONCE_NOW, + ConcurrencyPolicy.DISALLOW, Instant.parse("2099-01-01T00:00:00Z")); + ScheduleDefinition healthy = onceDefinition( + "healthy-during-retry", MisfirePolicy.FIRE_ONCE_NOW, + ConcurrencyPolicy.DISALLOW, Instant.parse("2099-01-01T00:00:00Z")); + service.create(retry); + service.create(healthy); + + service.triggerNow(retry.id(), "retry-1", Map.of()); + assertTrue(retryScheduled.await(5, TimeUnit.SECONDS)); + service.triggerNow(healthy.id(), "healthy-1", Map.of()); + + assertTrue("single Quartz worker remained occupied by delayed retry", + healthyCompleted.await(750, TimeUnit.MILLISECONDS)); + assertTrue(retryCompleted.await(5, TimeUnit.SECONDS)); + } + + /** 已持久化 fire 的重试不得因 replace 而切换到新 Handler。 */ + @Test + public void shouldKeepOriginalHandlerWhenScheduleIsReplacedDuringRetry() throws Exception { + AtomicInteger oldAttempts = new AtomicInteger(); + AtomicInteger newAttempts = new AtomicInteger(); + CountDownLatch oldRetryCompleted = new CountDownLatch(1); + service = newRamServiceWithHandlers(List.of( + handler("old-handler", context -> { + if (oldAttempts.incrementAndGet() == 1) { + throw new ScheduleRefireException("temporary registration failure", + new IllegalStateException("database unavailable"), 1, + Duration.ofMillis(400)); + } + oldRetryCompleted.countDown(); + }), + handler("new-handler", context -> newAttempts.incrementAndGet()) + ), 30_000L, 2); + ScheduleId id = new ScheduleId("test", "replace-during-retry"); + ScheduleDefinition original = definition(id, "old-handler"); + service.create(original); + service.triggerNow(id, "replace-retry-1", Map.of()); + awaitPersistentRetryTrigger(); + + service.replace(definition(id, "new-handler")); + + assertTrue("old handler retry did not complete", + oldRetryCompleted.await(5, TimeUnit.SECONDS)); + assertEquals(2, oldAttempts.get()); + assertEquals(0, newAttempts.get()); + } + /** * 验证立即触发会在返回回执前校验基础参数与覆盖参数的合并结果。 */ @@ -385,6 +492,23 @@ public class QuartzScheduleServiceTest { private QuartzScheduleService newRamService( Consumer handler, long shutdownWaitTimeoutMillis + ) throws Exception { + return newRamService(handler, shutdownWaitTimeoutMillis, 2); + } + + private QuartzScheduleService newRamService( + Consumer handler, + long shutdownWaitTimeoutMillis, + int threadCount + ) throws Exception { + return newRamServiceWithHandlers(List.of(handler("handler", handler)), + shutdownWaitTimeoutMillis, threadCount); + } + + private QuartzScheduleService newRamServiceWithHandlers( + List handlers, + long shutdownWaitTimeoutMillis, + int threadCount ) throws Exception { Properties properties = new Properties(); properties.setProperty( @@ -394,7 +518,7 @@ public class QuartzScheduleServiceTest { properties.setProperty("org.quartz.scheduler.instanceId", "NON_CLUSTERED"); properties.setProperty("org.quartz.scheduler.interruptJobsOnShutdown", "true"); properties.setProperty("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool"); - properties.setProperty("org.quartz.threadPool.threadCount", "2"); + properties.setProperty("org.quartz.threadPool.threadCount", Integer.toString(threadCount)); properties.setProperty("org.quartz.jobStore.class", "org.quartz.simpl.RAMJobStore"); properties.setProperty("org.quartz.jobStore.misfireThreshold", "100"); Scheduler scheduler = new StdSchedulerFactory(properties).getScheduler(); @@ -402,17 +526,7 @@ public class QuartzScheduleServiceTest { scheduler, true, shutdownWaitTimeoutMillis, - java.util.List.of(new ScheduleHandler() { - @Override - public String code() { - return "handler"; - } - - @Override - public void execute(ScheduleFireContext context) { - handler.accept(context); - } - }), + handlers, java.util.List.of() ); result.start(); @@ -420,6 +534,50 @@ public class QuartzScheduleServiceTest { return result; } + private void awaitPersistentRetryTrigger() throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (System.nanoTime() < deadline) { + boolean found = service.quartzScheduler() + .getTriggerKeys(GroupMatcher.anyTriggerGroup()) + .stream() + .anyMatch(key -> key.getGroup().startsWith( + QuartzScheduleMapper.GROUP_PREFIX + "retry.")); + if (found) return; + Thread.sleep(10L); + } + fail("persistent retry trigger was not created"); + } + + private static ScheduleHandler handler( + String code, + Consumer consumer + ) { + return new ScheduleHandler() { + @Override + public String code() { + return code; + } + + @Override + public void execute(ScheduleFireContext context) { + consumer.accept(context); + } + }; + } + + private static ScheduleDefinition definition(ScheduleId id, String handlerCode) { + return new ScheduleDefinition( + id, + handlerCode, + new OnceSchedulePlan(Instant.parse("2099-01-01T00:00:00Z")), + MisfirePolicy.FIRE_ONCE_NOW, + ConcurrencyPolicy.DISALLOW, + true, + Map.of(), + id.name() + ); + } + private static ScheduleDefinition cronDefinition(String handlerCode, String expression) { return new ScheduleDefinition( new ScheduleId("test", "lifecycle"), diff --git a/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactoryTest.java b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactoryTest.java new file mode 100644 index 0000000..664bb46 --- /dev/null +++ b/easy-agents-scheduler/easy-agents-scheduler-quartz/src/test/java/com/easyagents/scheduler/quartz/QuartzSchedulerFactoryTest.java @@ -0,0 +1,40 @@ +package com.easyagents.scheduler.quartz; + +import org.junit.Test; + +import java.util.Properties; + +import static org.junit.Assert.assertEquals; + +/** {@link QuartzSchedulerFactory} 原生属性映射测试。 */ +public class QuartzSchedulerFactoryTest { + + @Test + public void batchAcquisitionMustRunWithinJobStoreLock() { + QuartzSchedulerConfig config = new QuartzSchedulerConfig( + "batch-scheduler", + "NON_CLUSTERED", + "QRTZ_", + QuartzSchedulerConfig.STANDARD_JDBC_DELEGATE, + false, + 8, + Thread.NORM_PRIORITY, + 8, + 1_000L, + 15_000L, + 60_000L, + true, + 30_000L, + false + ); + + Properties properties = QuartzSchedulerFactory.properties(config, "testDs"); + + assertEquals("8", properties.getProperty( + "org.quartz.scheduler.batchTriggerAcquisitionMaxCount")); + assertEquals("1000", properties.getProperty( + "org.quartz.scheduler.batchTriggerAcquisitionFireAheadTimeWindow")); + assertEquals("true", properties.getProperty( + "org.quartz.jobStore.acquireTriggersWithinLock")); + } +} diff --git a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml index 2054a77..6c72a5e 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml +++ b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/pom.xml @@ -53,6 +53,11 @@ h2 test + + org.springframework + spring-jdbc + test + junit junit diff --git a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java index c12eb80..f6c9cf4 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java +++ b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfiguration.java @@ -18,6 +18,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.sql.init.dependency.DependsOnDatabaseInitialization; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -61,6 +62,7 @@ public class EasyAgentsSchedulerAutoConfiguration { */ @Bean(name = SCHEDULER_BEAN_NAME, initMethod = "start", destroyMethod = "close") @ConditionalOnMissingBean(ScheduleService.class) + @DependsOnDatabaseInitialization public QuartzScheduleService easyAgentsQuartzScheduleService( EasyAgentsSchedulerProperties properties, ListableBeanFactory beanFactory, @@ -83,6 +85,8 @@ public class EasyAgentsSchedulerAutoConfiguration { quartz.isClustered(), quartz.getThreadCount(), quartz.getThreadPriority(), + quartz.getBatchTriggerAcquisitionMaxCount(), + quartz.getBatchTriggerAcquisitionFireAheadTimeWindowMillis(), quartz.getClusterCheckinIntervalMillis(), quartz.getMisfireThresholdMillis(), quartz.isWaitForJobsToCompleteOnShutdown(), diff --git a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java index 0d1d055..c527309 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java +++ b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/main/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerProperties.java @@ -104,6 +104,12 @@ public class EasyAgentsSchedulerProperties { /** Quartz Worker 线程优先级。 */ private int threadPriority = Thread.NORM_PRIORITY; + /** 单次批量获取 Trigger 的最大数量。 */ + private int batchTriggerAcquisitionMaxCount = 1; + + /** 可提前纳入批量的时间窗口，单位毫秒。 */ + private long batchTriggerAcquisitionFireAheadTimeWindowMillis; + /** 集群心跳间隔，单位毫秒。 */ private long clusterCheckinIntervalMillis = 15_000L; @@ -251,6 +257,45 @@ public class EasyAgentsSchedulerProperties { this.threadPriority = threadPriority; } + /** + * 返回单次批量获取 Trigger 的最大数量。 + * + * @return 批量上限 + */ + public int getBatchTriggerAcquisitionMaxCount() { + return batchTriggerAcquisitionMaxCount; + } + + /** + * 设置单次批量获取 Trigger 的最大数量。 + * + * @param batchTriggerAcquisitionMaxCount 批量上限 + */ + public void setBatchTriggerAcquisitionMaxCount( + int batchTriggerAcquisitionMaxCount) { + this.batchTriggerAcquisitionMaxCount = batchTriggerAcquisitionMaxCount; + } + + /** + * 返回可提前纳入批量的时间窗口。 + * + * @return 毫秒窗口 + */ + public long getBatchTriggerAcquisitionFireAheadTimeWindowMillis() { + return batchTriggerAcquisitionFireAheadTimeWindowMillis; + } + + /** + * 设置可提前纳入批量的时间窗口。 + * + * @param batchTriggerAcquisitionFireAheadTimeWindowMillis 毫秒窗口 + */ + public void setBatchTriggerAcquisitionFireAheadTimeWindowMillis( + long batchTriggerAcquisitionFireAheadTimeWindowMillis) { + this.batchTriggerAcquisitionFireAheadTimeWindowMillis = + batchTriggerAcquisitionFireAheadTimeWindowMillis; + } + /** * 返回集群心跳间隔。 * diff --git a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java index debcfef..4e52025 100644 --- a/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java +++ b/easy-agents-scheduler/easy-agents-scheduler-spring-boot-starter/src/test/java/com/easyagents/scheduler/spring/boot/EasyAgentsSchedulerAutoConfigurationTest.java @@ -114,6 +114,47 @@ public class EasyAgentsSchedulerAutoConfigurationTest { } } + /** + * 验证调度器等待 Spring Boot 数据库脚本初始化完成后再启动。 + */ + @Test + public void shouldWaitForDatabaseInitializationBeforeStartingScheduler() throws Exception { + JdbcDataSource dataSource = dataSource(); + Map properties = enabledProperties(); + properties.put("easy-agents.scheduler.data-source-bean-name", "schedulerDataSource"); + properties.put("spring.sql.init.mode", "always"); + properties.put( + "spring.sql.init.schema-locations", + "classpath:quartz-schema/h2-2.5.2.sql" + ); + + SpringApplication application = new SpringApplication(AutoDiscoveryApplication.class); + application.setWebApplicationType(WebApplicationType.NONE); + application.setDefaultProperties(properties); + application.addInitializers(applicationContext -> { + GenericApplicationContext genericContext = + (GenericApplicationContext) applicationContext; + genericContext.registerBean( + "schedulerDataSource", + DataSource.class, + () -> dataSource + ); + }); + + try (ConfigurableApplicationContext context = application.run()) { + assertNotNull(context.getBean(ScheduleService.class)); + try ( + Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery( + "SELECT COUNT(*) FROM QRTZ_SCHEDULER_STATE" + ) + ) { + assertTrue(resultSet.next()); + } + } + } + /** * 验证多个 DataSource 未明确选择时启动失败并提供可操作信息。 */ @@ -162,6 +203,13 @@ public class EasyAgentsSchedulerAutoConfigurationTest { properties.put("easy-agents.scheduler.quartz.instance-id", "NON_CLUSTERED"); properties.put("easy-agents.scheduler.quartz.clustered", "false"); properties.put("easy-agents.scheduler.quartz.thread-count", "2"); + properties.put( + "easy-agents.scheduler.quartz.batch-trigger-acquisition-max-count", "2" + ); + properties.put( + "easy-agents.scheduler.quartz.batch-trigger-acquisition-fire-ahead-time-window-millis", + "500" + ); properties.put("easy-agents.scheduler.quartz.misfire-threshold-millis", "1000"); return properties; }