This is an automated email from the ASF dual-hosted git repository. jdaugherty pushed a commit to branch 8.0.x in repository https://gitbox.apache.org/repos/asf/grails-core.git
commit 8d373c178cd8e3d47b06c7c318ca6a552686caba Author: James Daugherty <[email protected]> AuthorDate: Sun Aug 23 00:04:09 2026 -0400 Document / add work around for long running jobs --- .../quartzLongRunningJobs.adoc | 157 ++++++++++++++++ grails-doc/src/en/guide/toc.yml | 1 + .../grails/plugins/quartz/GrailsJobFactory.java | 7 +- .../quartz/CustomTriggerFactoryBeanSpec.groovy | 50 ++++++ .../plugins/quartz/GrailsJobFactorySpec.groovy | 198 +++++++++++++++++++++ .../jobs/quartzapp/InterruptibleJob.groovy | 52 ++++++ .../jobs/quartzapp/LongRunningJob.groovy | 46 +++++ .../quartzapp/QuartzLongRunningJobSpec.groovy | 110 ++++++++++++ 8 files changed, 620 insertions(+), 1 deletion(-) diff --git a/grails-doc/src/en/guide/backgroundJobs/backgroundJobsAdvanced/quartzLongRunningJobs.adoc b/grails-doc/src/en/guide/backgroundJobs/backgroundJobsAdvanced/quartzLongRunningJobs.adoc new file mode 100644 index 0000000000..7004448251 --- /dev/null +++ b/grails-doc/src/en/guide/backgroundJobs/backgroundJobsAdvanced/quartzLongRunningJobs.adoc @@ -0,0 +1,157 @@ +//// +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you 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. +//// + +A job whose execution takes longer than the interval of the trigger that started it needs three decisions: whether its executions may overlap, how many of the scheduler's threads they may occupy, and how an execution that has run for too long is stopped. + +===== Overlapping executions + +`concurrent` defaults to `true`, so every fire starts an execution whether or not the previous one has finished. A job that runs for hours on a trigger that fires every few minutes therefore stacks executions up, each holding one of the scheduler's worker threads for its whole run. Once every thread is occupied nothing fires at all: the scheduler looks stuck, the log fills with misfires, and no work gets done. + +Set `concurrent = false` on such a job, and the scheduler never runs two of its executions at the same time: + +[source,groovy] +---- +class ImportJob { + + static concurrent = false + + static triggers = { + simple repeatInterval: 100_000L + } + + void execute() { + // ... + } +} +---- + +A fire that arrives while the job is still running is then held, and what becomes of it is the trigger's misfire instruction. + +===== Threads + +Jobs run on a fixed pool of worker threads — ten of them unless configured otherwise. A long-running job holds one of those threads for its entire execution, so the pool has to be large enough for every execution that can be in flight at once, plus the short jobs that must keep firing meanwhile: + +[source,yaml] +---- +quartz: + threadPool: + threadCount: 25 +---- + +===== Misfires + +A trigger misfires when the scheduler cannot fire it when it was due — because the previous execution is still running and the job is not concurrent, or because no worker thread is free. What happens to the fires that were missed is the trigger's `misfireInstruction`, set like any other trigger attribute. Left unset, a trigger uses Quartz's smart policy, which depends on the trigger type: an indefinitely repeating `simple` trigger drops the fires it missed and carries on at its next sche [...] + +So to have an hourly report skip the run it missed instead of starting it the moment the long execution finishes: + +[source,groovy] +---- +class ReportJob { + + static concurrent = false + + static triggers = { + cron cronExpression: '0 0 * * * ?', + misfireInstruction: CronTrigger.MISFIRE_INSTRUCTION_DO_NOTHING + } + + void execute() { + // ... + } +} +---- + +The instructions to choose from are the constants of the Quartz trigger type — `SimpleTrigger` and `CronTrigger`. How late a fire may be before the scheduler counts it as a misfire is `quartz.jobStore.misfireThreshold`, in milliseconds; it is passed straight through to Quartz, so set it explicitly if the tolerance matters to your schedule. + +===== Watching what is running + +`jobManagerService.runningJobs` returns the `JobExecutionContext` of every execution in flight. Each one carries the job it belongs to as `jobDetail.key` and the moment its execution began as `fireTime`, which is all that is needed to find work that has been running for too long: + +[source,groovy] +---- +jobManagerService.runningJobs.each { JobExecutionContext context -> + println "${context.jobDetail.key} has been running since ${context.fireTime}" +} +---- + +===== Stopping an execution + +There is no timeout setting. Nothing can safely stop a thread in the middle of arbitrary work, so a job is asked to stop itself instead: declare an `interrupt()` method on the job, and the plugin routes interruption requests to it. It is up to the job to notice and return from `execute`. + +[source,groovy] +---- +class ImportJob { + + static concurrent = false + + private volatile boolean interrupted + + static triggers = { + simple repeatInterval: 100_000L + } + + void execute() { + for (batch in batches) { + if (interrupted) { + return + } + process batch + } + } + + void interrupt() { + interrupted = true + } +} +---- + +A field is enough to carry the flag: every execution is given its own instance of the job class, so nothing is shared between executions. + +Ask for the interruption from anywhere in the application, by job group and job name: + +[source,groovy] +---- +jobManagerService.interruptJob('GRAILS_JOBS', 'com.example.ImportJob') +---- + +A job that does not declare `interrupt()` cannot be stopped; the request raises an `UnableToInterruptJobException` saying so. Code that interrupts jobs it does not own — a watchdog job which interrupts whatever it finds running past a limit, for instance — has to allow for that. + +===== Hibernate sessions + +A Hibernate `Session` is bound to the job's thread for the whole execution unless the job sets `sessionRequired = false`. A session that lives for hours keeps every entity it has loaded in its first-level cache, so memory grows with the work done and the entities it holds drift further from the database. + +Process the work of a long job in chunks, each in a session of its own, so neither grows without bound: + +[source,groovy] +---- +class ImportJob { + + static concurrent = false + + void execute() { + batches.each { batch -> + Invoice.withNewSession { + process batch + } + } + } +} +---- + +Note that the session is bound regardless of `sessionRequired` when `quartz.jdbcStore` is enabled, because the job store needs it. diff --git a/grails-doc/src/en/guide/toc.yml b/grails-doc/src/en/guide/toc.yml index 611ddbc9ac..547205767f 100644 --- a/grails-doc/src/en/guide/toc.yml +++ b/grails-doc/src/en/guide/toc.yml @@ -356,6 +356,7 @@ backgroundJobs: quartzJobProperties: Job Properties quartzDynamicScheduling: Dynamic Job Scheduling quartzJobManagerService: Managing Jobs at Runtime + quartzLongRunningJobs: Long-Running Jobs quartzConfiguration: Configuration quartzClustering: Clustering quartzUsefulLinks: Useful Links diff --git a/grails-quartz/src/main/groovy/grails/plugins/quartz/GrailsJobFactory.java b/grails-quartz/src/main/groovy/grails/plugins/quartz/GrailsJobFactory.java index 4b46012745..22dc00b6b1 100644 --- a/grails-quartz/src/main/groovy/grails/plugins/quartz/GrailsJobFactory.java +++ b/grails-quartz/src/main/groovy/grails/plugins/quartz/GrailsJobFactory.java @@ -138,7 +138,12 @@ public class GrailsJobFactory extends AdaptableJobFactory implements Application throw new UnableToInterruptJobException(e); } } else { - throw new UnableToInterruptJobException(job.getClass().getName() + " doesn't support interruption"); + throw new UnableToInterruptJobException( + MessageFormat.format( + "{0} does not declare an {1}() method, so it cannot be interrupted", + job.getClass().getName(), GrailsJobClassConstants.INTERRUPT + ) + ); } } diff --git a/grails-quartz/src/test/groovy/grails/plugins/quartz/CustomTriggerFactoryBeanSpec.groovy b/grails-quartz/src/test/groovy/grails/plugins/quartz/CustomTriggerFactoryBeanSpec.groovy index 31c977a66d..854a1dbeb4 100644 --- a/grails-quartz/src/test/groovy/grails/plugins/quartz/CustomTriggerFactoryBeanSpec.groovy +++ b/grails-quartz/src/test/groovy/grails/plugins/quartz/CustomTriggerFactoryBeanSpec.groovy @@ -87,4 +87,54 @@ class CustomTriggerFactoryBeanSpec extends Specification { assert DateBuilder.IntervalUnit.MINUTE == customTrigger.repeatIntervalUnit assert 5 == customTrigger.repeatInterval } + + void 'the misfire instruction a trigger declares is carried by the trigger'() { + setup: + def builder = new TriggersConfigBuilder('TestJob', null) + builder.build { + simple name: 'simple', repeatInterval: 1000, + misfireInstruction: SimpleTrigger.MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT + cron name: 'cron', cronExpression: CRON_EXPRESSION, + misfireInstruction: CronTrigger.MISFIRE_INSTRUCTION_DO_NOTHING + } + + Map<String, Trigger> triggers = [:] + + builder.triggers.values().each { + CustomTriggerFactoryBean factory = new CustomTriggerFactoryBean() + factory.setTriggerClass(it.triggerClass) + factory.setTriggerAttributes(it.triggerAttributes) + factory.afterPropertiesSet() + Trigger trigger = factory.getObject() as Trigger + triggers.put(trigger.key.name, trigger) + } + + expect: + triggers['simple'].misfireInstruction == SimpleTrigger.MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT + triggers['cron'].misfireInstruction == CronTrigger.MISFIRE_INSTRUCTION_DO_NOTHING + } + + void 'a trigger uses the smart misfire policy of its type unless it declares an instruction'() { + setup: + def builder = new TriggersConfigBuilder('TestJob', null) + builder.build { + simple name: 'simple', repeatInterval: 1000 + cron name: 'cron', cronExpression: CRON_EXPRESSION + } + + Map<String, Trigger> triggers = [:] + + builder.triggers.values().each { + CustomTriggerFactoryBean factory = new CustomTriggerFactoryBean() + factory.setTriggerClass(it.triggerClass) + factory.setTriggerAttributes(it.triggerAttributes) + factory.afterPropertiesSet() + Trigger trigger = factory.getObject() as Trigger + triggers.put(trigger.key.name, trigger) + } + + expect: + triggers['simple'].misfireInstruction == Trigger.MISFIRE_INSTRUCTION_SMART_POLICY + triggers['cron'].misfireInstruction == Trigger.MISFIRE_INSTRUCTION_SMART_POLICY + } } diff --git a/grails-quartz/src/test/groovy/grails/plugins/quartz/GrailsJobFactorySpec.groovy b/grails-quartz/src/test/groovy/grails/plugins/quartz/GrailsJobFactorySpec.groovy new file mode 100644 index 0000000000..4402e9beba --- /dev/null +++ b/grails-quartz/src/test/groovy/grails/plugins/quartz/GrailsJobFactorySpec.groovy @@ -0,0 +1,198 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 grails.plugins.quartz + +import org.quartz.Job +import org.quartz.JobDetail +import org.quartz.JobExecutionContext +import org.quartz.JobExecutionException +import org.quartz.UnableToInterruptJobException +import org.quartz.impl.triggers.SimpleTriggerImpl +import org.quartz.spi.TriggerFiredBundle +import org.springframework.context.support.StaticApplicationContext +import spock.lang.Specification + +/** + * Tests the job the factory hands the scheduler for each execution of a job artefact. + */ +class GrailsJobFactorySpec extends Specification { + + private static final String JOB_NAME = 'grails.plugins.quartz.TestJob' + private static final String JOB_GROUP = 'GRAILS_JOBS' + + void 'the job the factory creates runs the execute method of the job artefact'() { + given: + SimpleJob artefact = new SimpleJob() + + when: + Job job = jobFor(artefact) + job.execute(Mock(JobExecutionContext)) + + then: + artefact.executions == 1 + } + + void 'a job artefact whose execute method takes the execution context is given it'() { + given: + ContextAwareJob artefact = new ContextAwareJob() + JobExecutionContext context = Mock(JobExecutionContext) + + when: + jobFor(artefact).execute(context) + + then: + artefact.context.is(context) + } + + void 'a job artefact without an execute method is rejected'() { + when: + new GrailsJobFactory.GrailsJob(new NoExecuteJob()) + + then: + IllegalArgumentException e = thrown() + e.message.contains(NoExecuteJob.name) + e.message.contains('execute') + } + + void 'a job artefact whose execute method takes more than one argument is rejected'() { + when: + new GrailsJobFactory.GrailsJob(new TooManyArgumentsJob()) + + then: + IllegalArgumentException e = thrown() + e.message.contains(TooManyArgumentsJob.name) + } + + void 'an exception thrown by a job artefact is reported as a job execution exception'() { + when: + jobFor(new FailingJob()).execute(Mock(JobExecutionContext)) + + then: + JobExecutionException e = thrown() + e.cause instanceof IllegalStateException + e.cause.message == 'the job failed' + } + + void 'a job execution exception thrown by a job artefact is reported as it is'() { + given: + JobExecutionException thrownByJob = new JobExecutionException('unschedule me') + + when: + jobFor(new JobExecutionExceptionJob(exception: thrownByJob)).execute(Mock(JobExecutionContext)) + + then: + JobExecutionException e = thrown() + e.is(thrownByJob) + } + + void 'interrupting a job calls the interrupt method of the job artefact'() { + given: + InterruptibleJob artefact = new InterruptibleJob() + Job job = jobFor(artefact) + + when: + job.interrupt() + + then: + artefact.interrupted + } + + void 'interrupting a job artefact which does not declare an interrupt method reports that it cannot be interrupted'() { + given: + Job job = jobFor(new SimpleJob()) + + when: + job.interrupt() + + then: + UnableToInterruptJobException e = thrown() + e.message == "${SimpleJob.name} does not declare an interrupt() method, so it cannot be interrupted" + } + + /** + * Builds the job the way the scheduler does: through the factory, which looks the job artefact up + * in the application context by the name the job detail carries. + */ + private Job jobFor(Object artefact) { + StaticApplicationContext applicationContext = new StaticApplicationContext() + applicationContext.beanFactory.registerSingleton(JOB_NAME, artefact) + applicationContext.refresh() + + GrailsJobFactory factory = new GrailsJobFactory() + factory.applicationContext = applicationContext + + factory.newJob(new TriggerFiredBundle(jobDetail(), new SimpleTriggerImpl('trigger', 'GRAILS_TRIGGERS'), + null, false, new Date(), null, null, null), null) + } + + private JobDetail jobDetail() { + JobDetailFactoryBean factory = new JobDetailFactoryBean() + factory.jobClass = new GrailsJobClassMock(fullName: JOB_NAME, group: JOB_GROUP, concurrent: true) + factory.afterPropertiesSet() + factory.object + } +} + +class SimpleJob { + int executions + + void execute() { + executions++ + } +} + +class ContextAwareJob { + JobExecutionContext context + + void execute(JobExecutionContext context) { + this.context = context + } +} + +class NoExecuteJob { +} + +class TooManyArgumentsJob { + void execute(JobExecutionContext context, String other) {} +} + +class FailingJob { + void execute() { + throw new IllegalStateException('the job failed') + } +} + +class JobExecutionExceptionJob { + JobExecutionException exception + + void execute() { + throw exception + } +} + +class InterruptibleJob { + boolean interrupted + + void execute() {} + + void interrupt() { + interrupted = true + } +} diff --git a/grails-test-examples/quartz/grails-app/jobs/quartzapp/InterruptibleJob.groovy b/grails-test-examples/quartz/grails-app/jobs/quartzapp/InterruptibleJob.groovy new file mode 100644 index 0000000000..348b2cc568 --- /dev/null +++ b/grails-test-examples/quartz/grails-app/jobs/quartzapp/InterruptibleJob.groovy @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 quartzapp + +import java.util.concurrent.atomic.AtomicInteger + +/** + * A job which keeps working until it is interrupted, the way an application stops work that has been + * running for too long. It has no triggers, so it only runs when a test triggers it. + */ +class InterruptibleJob { + + static final AtomicInteger STARTED = new AtomicInteger() + static final AtomicInteger INTERRUPTED = new AtomicInteger() + + private volatile boolean interrupted + + static triggers = { + } + + void execute() { + STARTED.incrementAndGet() + long deadline = System.currentTimeMillis() + 30_000 + while (!interrupted && System.currentTimeMillis() < deadline) { + sleep 50 + } + if (interrupted) { + INTERRUPTED.incrementAndGet() + } + } + + void interrupt() { + interrupted = true + } +} diff --git a/grails-test-examples/quartz/grails-app/jobs/quartzapp/LongRunningJob.groovy b/grails-test-examples/quartz/grails-app/jobs/quartzapp/LongRunningJob.groovy new file mode 100644 index 0000000000..609f73a3b3 --- /dev/null +++ b/grails-test-examples/quartz/grails-app/jobs/quartzapp/LongRunningJob.groovy @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 quartzapp + +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger + +/** + * A job which runs until it is released, so that a test can observe what the scheduler does while an + * execution is still in flight. It has no triggers, so it only runs when a test triggers it. + */ +class LongRunningJob { + + static concurrent = false + + static final AtomicInteger STARTED = new AtomicInteger() + static final AtomicInteger FINISHED = new AtomicInteger() + static final CountDownLatch RELEASE = new CountDownLatch(1) + + static triggers = { + } + + void execute() { + STARTED.incrementAndGet() + RELEASE.await(30, TimeUnit.SECONDS) + FINISHED.incrementAndGet() + } +} diff --git a/grails-test-examples/quartz/src/integration-test/groovy/quartzapp/QuartzLongRunningJobSpec.groovy b/grails-test-examples/quartz/src/integration-test/groovy/quartzapp/QuartzLongRunningJobSpec.groovy new file mode 100644 index 0000000000..15a5ea3d4f --- /dev/null +++ b/grails-test-examples/quartz/src/integration-test/groovy/quartzapp/QuartzLongRunningJobSpec.groovy @@ -0,0 +1,110 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 quartzapp + +import grails.plugins.quartz.JobManagerService +import grails.testing.mixin.integration.Integration +import org.quartz.JobExecutionContext +import org.springframework.beans.factory.annotation.Autowired +import spock.lang.Specification +import spock.util.concurrent.PollingConditions + +/** + * Verifies what a running application does with a job whose execution outlives the fire that started it: + * executions are serialized when the job asks for it, and a running execution can be observed and stopped. + */ +@Integration(applicationClass = Application) +class QuartzLongRunningJobSpec extends Specification { + + private static final String JOBS_GROUP = 'GRAILS_JOBS' + + @Autowired + JobManagerService jobManagerService + + void cleanup() { + LongRunningJob.RELEASE.countDown() + } + + void 'a job which is not concurrent is not executed while an earlier execution is still running'() { + given: + PollingConditions conditions = new PollingConditions(timeout: 30) + + when: 'a job that runs until it is released is triggered twice' + LongRunningJob.triggerNow() + + then: + conditions.eventually { + assert LongRunningJob.STARTED.get() == 1 + } + + and: 'the execution is reported as running, with the moment it started' + jobManagerService.runningJobs.any { JobExecutionContext context -> + context.jobDetail.key.name == LongRunningJob.name && context.fireTime != null + } + + when: 'the application tries to stop an execution of a job which cannot be interrupted' + jobManagerService.interruptJob(JOBS_GROUP, LongRunningJob.name) + + then: 'it is told that the job does not support it' + Exception e = thrown() + e.message.contains('does not declare an interrupt() method') + + when: 'a second fire arrives while the first execution is still in flight' + LongRunningJob.triggerNow() + + then: 'it waits rather than running a second execution alongside the first' + conditions.within(2) { + assert LongRunningJob.STARTED.get() == 1 + } + + when: 'the first execution finishes' + LongRunningJob.RELEASE.countDown() + + then: 'the fire that was held runs' + conditions.eventually { + assert LongRunningJob.STARTED.get() == 2 + assert LongRunningJob.FINISHED.get() == 2 + } + } + + void 'a job that runs for too long is stopped by interrupting it'() { + given: + PollingConditions conditions = new PollingConditions(timeout: 30) + + when: + InterruptibleJob.triggerNow() + + then: + conditions.eventually { + assert InterruptibleJob.STARTED.get() == 1 + } + + when: 'the application interrupts the job it finds running' + jobManagerService.interruptJob(JOBS_GROUP, InterruptibleJob.name) + + then: 'the execution stops, and the scheduler is left with nothing running' + conditions.eventually { + assert InterruptibleJob.INTERRUPTED.get() == 1 + assert jobManagerService.runningJobs.every { JobExecutionContext context -> + context.jobDetail.key.name != InterruptibleJob.name + } + } + } +}
