[EPIC]: Engine Abstraction

[Zelldon]

Why?

Our team has grown to a certain size, which makes it hard to manage and keep being focused (e.g. discussions etc.). We will split our development team into two separate teams to keep being focused on certain topics/issues, and the team and discussions around important topics manageable.

The idea is the following:

The Zeebe Distributed Platform (ZDP) Team provides an abstraction layer around how records are stored/persisted, read, and written. The Zeebe Platform Automation (ZPA) Team is in charge of processing the records and updating their state. With that, the idea is that both teams can fully concentrate/focus on their topics, e.g. ZPA on the real business logic.

In order to support this split, we will need some better abstraction between certain layers/modules/parts of the system. One of them is the engine.

Todo:

Design / Finding phase

• Create a proposal for a possible abstraction between both teams for the engine
  • Write down an proposal, and validate with the team Engine Abstraction proposal #9601
  • Do a proof of concept of that proposal [POC]: Engine Abstraction #9602
• Discuss the POC Results [POC]: Engine Abstraction #9602 (comment)
• If proposal and POC successful completed, do a further break down of following tasks.

Implementation:

^{I marked the path blue which is necessary to satisfy #9420. This means we need to have the error handling in the engine implementation in order to achieve this.}

• Remove LogStream writers from Engine #9724

  • In order to work on this in parallel we copy the TypedStreamWriterImpl to a new class and make certain changes
    • Copy the content from TypedStreamWriterImpl to a new class
    • Remove the LogStreamBatch usage from that class, write directly into a pre-claimed buffer
    • Add a new method to the LogStreamBatch
    • Write tests for both
  • Find a good new name for the Writers Interface, including Command-, Rejection-, StateWriter. In the POC [POC]: Engine Abstraction #9602, we called it Builders, because wanted in the end build a list of Records. But maybe we can also use a different name here.
  • Create a new Result Interface / Class which can be returned by the Engine
    • The Result can for simplicity return a BufferWriter like we did in the POC [POC]: Engine Abstraction #9602 (we can improve that later), but we can also just return a list of Records if you find a good way for it
    • Result should be used by Processing and ScheduledTasks to return Results
  • Engine should return the result - ⚠️ here we need the Stream Processing changes first
  • Use result in platform, write the records to the LogStreamBatchWriter
  • Bonus: The result contains only the list of the records (instead of the general Buffer or BufferWriter)
  • Bonus2 The serialization is done in the StreamPlatform on writing to the LogStreamBatchWriter
  • We might need to adjust some tests

• Refactor StreamProcessor / Engine #9725

  • Mark the processing explicitly as currently ongoing, via flag. The currentProcessor field shouldn't be used anymore.
  • Find a good name for the StreamProcessor
  • Rename the current StreamProcessor, e.g. StreamingPlatform (see [POC]: Engine Abstraction #9602)
  • Introduce a new StreamProcessor interface, which covers the reality (this needs to be adjusted during progressing with [EPIC]: Engine Abstraction #9600 )
    • init Method
    • replay Method
    • process Method (at the beginning with no Result returning, only after parts of Remove LogStream writers from Engine #9724 are done)
    • onError Method - might be empty at the begin
    • Implement LifecycleAware?
  • Create a new StreamProcessor implementation called Engine
  • Move some code out from the StreamingPlatform, ReplayStateMachine and ProcessingStateMachine to the Engine, see POC [POC]: Engine Abstraction #9602 and related branch
  • ⚠️ OnError implementation, can be completed after the writer change is done and we can return a Result see Remove LogStream writers from Engine #9724
    • Transaction is rolled back by StreamPlatform, new Transaction is created and Engine is called (onError)
    • Engine should be able to receive the Throwable, Command and current position (?)
    • Based on the input it can return a Result, which will be written to the log, and the transaction is committed again. Here we can use the same path from the ProcessingStateMachine as we do from the normal processing.
  • Bonus: If we create the Engine outside of the StreamingPlatform and give it to the builder, we can inject the dependencies outside and can simplify the StreamProcessor tests
  • Cleanup
    • Remove dependency of engine package to StreamProcessorListener #10034

• Introduce ProcessingScheduleService #9730

  • 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

• Iterate over Side effects #9723

  • Clarify that use case or question whether we still want to support executing side effects after successful processing only Engine abstraction - Investigate side effects #9686
  • If yes, we might want to improve the current side effects to make it not so unstable as it is right now (if a side effect is added it is likely the previous is overwritten). We maybe want to support a queue
  • If no, remove the side effects and send requests/responses directly

• Clean up StreamingPlatform / Engine #9727

  • Remove the position from the TypedRecordProcessor#process see [POC]: Engine Abstraction #9602
  • Remove ProxyWriter from the ProcessingContext see [POC]: Engine Abstraction #9602
  • Remove proxy writers
  • Remove writers from the processing interface after Remove LogStream writers from Engine #9724 and Iterate over Side effects #9723
  • Clean up ProcessingContext, remove unused classes/objects
  • Bonus: Split up the engine module into Stream Platform and Engine and separate the components

• Update downstream repos

  • Zeebe process test
  • Eze