Introduce ProcessingScheduleService

[Zelldon]

Description

Part of #9600

Create an abstraction around the processing actor, since we want to remove the dependency and knowledge about the current execution model from the engine. This will simplify the tests as we can use a simple implementation for the execution and it should also help us to make further changes to the actor scheduler etc.

See SchedulingService in #9602 (comment)

Todo:

• Create a new interface for the ProcessingScheduleService (with narrowed scope)
  • Possibily only two methods, runDelayed and runComplete take a look at the POC [POC]: Engine Abstraction #9602
  • Ideally we use real futures on the runComplete, but might be something we refactor later
  • The interface contains a way to return some records, which will be written later on
  • Support runAfterCommit Iterate over Side effects #9723 (comment) to replace side effects
• Implement the interface, and provide certain guarantees, take a look at POC for more details
  • Guarantee 1: We are executed after processing and committing transaction (e.g. if scheduling happened during processing)
  • Guarantee 2: We are not executed in parallel with processing (discussable, depends on the Writers and Result implementation) We removed this again Run ScheduledTasks during processing #10355
• Bonus: We might want to execute multiple delayed tasks on parallel maybe even in parallel during processing, for that we need to make sure that resources are not shared
  • The above was not easily possible with Remove the extra submit from ProcessingScheduleServiceImpl #10428 we removed the shared writer, and the addition submit, but other than that it was not possible to create a new actor for it.
  • The consumer of the SchedulingService have only ReadOnly access to the database This was not easily possible
• Add unit tests for the implementation
• Before migrating to the new abstraction, migrate the ActorCOntrol#runAtFixedRate consumers to the #runDelayed usage, this means after each run the job needs to be scheduled again
• Integrate the ProcessingScheduler in the ProcessingStateMachine, means if task is executed we can write the result to the log ProcessingScheduleService allows to schedule a task which returns a Result #9999
• Migrate step by step the actorControl usage
• Remove the actor control from the ProcessingContext
• Remove LogStreamWriter from the ProcessingContext

[Zelldon]

Question raised from @npepinpe after the POC:

#9602 (comment)
Abstracting actor scheduler sounds good. What is the minimal interface? Can we get rid of runUntilDone and the likes, and just go for two capabilities: defer execution (e.g. run or submit) and schedule (e.g. runDelayed)? That would greatly simplify things. The caller can take care of runUntilDone by implementing their own state machine which ensures no other tasks can be completed in between. However if we want to keep the queuing mechanism, then this is not possible. Do we need it, though?

I think this is a bit unrelated to the topic and can be discussed with the PDT regarding the actor scheduler changes. The ProcessingScheduleService will have a really limited interface, which you can see above.

[npepinpe]

Do you mean this one?

interface SchedulingService {
  Collection<Records> schedule(Duration d, Runnable r);
}

[Zelldon]

No sorry. For example in the POC it looked like this: 176cfab#diff-f27601ec63a23e86b4a992a9cd4ac54192d81fbea9da126a8edeaa3b20123427R483-R487

public interface ProcessingSchedulingService {

   void runWithDelay(Duration duration, Supplier<ProcessingResult> scheduledProcessing);
 }

We need to pass in an Supplier otherwise it doesn't work, the method itself can't return this.

It depends now on #9724 how the result looks like, but this will be then the return of the Supplier or function or what ever is passed in.

[pihme]

This comment came up during a discussion: #9756 (review)

Regarding timers, we do return a cancellable ScheduledTimer, which will prevent the execution of the timer. I didn't notice it being used during cancellation. I know we need to do both (since if the timer was added to the actor task queue before it's cancelled, it will be executed regardless), but it may avoid unnecessary runs. If we don't want to expand the API with another interface just yet, we could return something like a CloseableSilently for now. It does mean keeping track of the current timer however

For now, I didn't follow that suggestion and want to see whether we can do without it.

[Zelldon]

With #10428 we completed this part of #9600