Refactor StreamProcessor / Engine

[Zelldon]

Description

Part of #9600

Rename StreamProcessor, to better distinguish between the Platform which drives the Processor, the StreamProcessor interface, and the Engine as implementation, etc. Furthermore, refactor the current implementation to make it possible to split up the modules and have clear boundaries between components.

See Processing in #9602 (comment)

Todo:

• 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

[Zelldon]

There was a question raised by @npepinpe regarding error handling in onError:

#9602 (comment)
What if an error occurs in onProcessingError? Do we have some escape hatch? How about UnrecoverableError classes? How > much control do we give to the engine (i.e. can it shut itself down on unrecoverable error)?

My answer would be similar like we do it now: https://github.com/camunda/zeebe/blob/main/engine/src/main/java/io/camunda/zeebe/streamprocessor/ProcessingStateMachine.java#L56-L87

 * +------------------+            +--------------------+
 * |                  |            |                    |      exception
 * | readNextRecord() |----------->|  processCommand()  |------------------+
 * |                  |            |                    |                  v
 * +------------------+            +--------------------+            +---------------+
 *           ^                             |                         |               |------+
 *           |                             |         +-------------->|   onError()   |      | exception
 *           |                             |         |  exception    |               |<-----+
 *           |                     +-------v-------------+           +---------------+
 *           |                     |                     |                 |
 *           |                     |   writeRecords()    |                 |
 *           |                     |                     |<----------------+
 * +----------------------+        +---------------------+
 * |                      |                 |
 * | executeSideEffects() |                 v
 * |                      |       +----------------------+
 * +----------------------+       |                      |
 *           ^                    |     updateState()    |
 *           +--------------------|                      |
 *                                +----------------------+
 *                                       ^      |
 *                                       |      | exception
 *                                       |      |
 *                                    +---------v----+
 *                                    |              |
 *                                    |   onError()  |
 *                                    |              |
 *                                    +--------------+
 *                                       ^     |
 *                                       |     |  exception
 *                                       +-----+
 *

[npepinpe]

To clarify, this means we loop in the engine?

[Zelldon]

No.

The engine is called with an error and record (onError). If here an exception occurs again then, we again catch that in the StreamProcessor, rollback the transaction and start new a transaction and call onError again.

[npepinpe]

Thanks! That's not obvious from the diagram, it looks like we loop in onError 😄

[Zelldon]

Yeah this state machine is the current state, and taken from the current StreamProcessor sorry 😅

[Zelldon]

The refactoring of the streamprocessor / engine interface is done. There might be some clean up which we will do with #9727