Support @Scheduled on Reactive methods and Kotlin suspending functions
#29924
Conversation
simonbasle
added
in: core
|
We should probably review #28515 at the same time and have a common approach. |
|
As discussed, let's implement the Coroutines support as part of this PR via an invocation of |
@sdeleuze done :) As stated in the latest commit, anything |
.../main/java/org/springframework/scheduling/annotation/ScheduledAnnotationReactiveSupport.java
Outdated
Show resolved
Hide resolved
.../main/java/org/springframework/scheduling/annotation/ScheduledAnnotationReactiveSupport.java
Show resolved
Hide resolved
...ain/java/org/springframework/scheduling/annotation/ScheduledAnnotationBeanPostProcessor.java
Show resolved
Hide resolved
...ain/java/org/springframework/scheduling/annotation/ScheduledAnnotationBeanPostProcessor.java
Outdated
Show resolved
Hide resolved
.../java/org/springframework/scheduling/annotation/ScheduledAnnotationReactiveSupportTests.java
Outdated
Show resolved
Hide resolved
.../java/org/springframework/scheduling/annotation/ScheduledAnnotationReactiveSupportTests.java
Outdated
Show resolved
Hide resolved
|
Thanks for tackling this - it's definitely the right time to address this as we'll work on observability in #29883. |
There was a problem hiding this comment.
It's good to see this planned.
My initial comment is to discuss the choice to use Reactor for scheduling. It means a more limited usage style, excluding cron expressions. Also, limits the usefulness of ReactorAdapterRegistry to a degree by requiring Reactor for the scheduling. I'm also wondering if ReactiveTask is going to cause classpath issues for sync methods but that's a more secondary, implementation concern.
Could the scheduling infrastructure used for sync also be used for reactive methods too, with some additional logic to subscribe to the Publisher every time a task is scheduled?
simonbasle
force-pushed
the
23533-scheduledForPublishers
branch
from
3b22604
to
038c8dd
Compare
|
In the few last commits, I have attempted to take that approach which reuses a maximum of the existing infrastructure by using |
This commit adds support for `@Scheduled` annotation on reactive methods in fixed delay or fixed rate mode. Cron mode is not supported. Reactive methods are methods that return a Publisher or a subclass of Publisher. This is only considered if Reactor (and Reactive Streams) is present at runtime. This is implemented using Reactor operators, as a `Flux<Void>` that repeatedly flatmaps to the `Publisher` in the background, re-subscribing to it according to the `@Scheduled` configuration. The method which creates the `Publisher` is only called once. If the `Publisher` errors, the exception is logged at warn level and otherwise ignored. As a result the underlying `Flux` will continue re-subscribing to the `Publisher`. Note that if Reactor is not present, the standard processing applies and the method itself will repeatedly be scheduled for execution, ignoring the returned `Publisher` and not even subscribing to it. This effectively becomes a no-op operation unless the method has other side effects. Closes spring-projectsgh-23533
The support's `isReactive` method checks Kotlin suspending functions first then reactive (Publisher-returning) methods second. It asserts the relevant runtime, all in a single call. Similarly, turning the Method into a Publisher is done via a single common helper method `getPublisherFor(Method, Object)`. All imports of reactive classes and other reactive-specific logic is still outsourced to ScheduledAnnotationReactiveSupport in order to avoid any classpath issue in the bean postprocessor.
|
I have now rebased on top of |
simonbasle
force-pushed
the
23533-scheduledForPublishers
branch
from
62f461d
to
5b3ae76
Compare
simonbasle
changed the title
@Scheduled on Reactive methods and Kotlin suspending functions
|
I'm currently working on the observability support of scheduled tasks in #29883. It looks like the current proposal for reactive methods support has a difference of behavior. Let's use two variants of scheduled methods, one blocking and the other reactive, both throwing exceptions during their processing: @Component
public class ScheduledComponent {
private static final Logger logger = LoggerFactory.getLogger(ScheduledComponent.class);
@Scheduled(cron = "0,10,20,30,40,50 * * * * *")
public void blocking() {
logger.info("Executing 'blocking' @Scheduled method");
throw new IllegalStateException("Blocking method failed");
}
@Scheduled(cron = "2,12,22,32,42,52 * * * * *")
public Mono<Void> reactive() {
return Mono.error(() -> new IllegalStateException("Reactive method failed"))
.doOnError(exc -> logger.info("Executing 'reactive' @Scheduled method"))
.then();
}
}The blocking variant throws the exception from the generated runnable, and it's caught by the With the current proposal for reactive support, exceptions are never thrown and the error handling contract is not used: I don't see this as a limitation, but really as a key difference between imperative and async in general. This behavior is already documented in this PR, but maybe we can mention that the |
|
Closed with 35052f2 |
This commit adds support for
@Scheduledannotation on reactive methodsand on Kotlin suspending functions (provided a bridge to Reactive Streams
is available at runtime).
Reactive methods are methods that return an instance of Publisher or an
instance of a class that the ReactiveAdapterRegistry can adapt to a
Publisher with deferred support.
Kotlin suspending functions are converted to Publisher as well, using
CoroutineUtilsand the support of thekotlinx.coroutines.reactorbridge.
The bean method that produces the
Publisheris only called once, butthere can be multiple subscriptions to that instance as configured by
the annotation. The usual task-based infrastructure for synchronous code
is reused for the purpose of scheduling the subscriptions.
One special case is when a
fixedDelayis used, as the non-blockingnature of subscribing to a Publisher makes it harder to adhere to these
semantics. As a result, for that particular case only, the subscription
is done in a blocking fashion inside the scheduled task.
Publisher onNext events are ignored. Active subscriptions are tracked
by the processor so that long-running publishers and infinite publishers
are also supported, allowing for cancellation if the associated bean is
destroyed or if the context is stopped.
See gh-23533
Closes gh-29924