refactor(engine): append side-effects to writer

This removes difficult to understand abstractions around side effects
and ensures that the engine does not re-use side-effect queues which
complicate batch processing and processing of uncommitted.

Adds a `SideEffectWriter` backed by the `ProcessingResultBuilder`
that is used throughout like other writers.

(cherry picked from commit 9eac8c3)

engine/src/main/java/io/camunda/zeebe/engine/Engine.java

@@ -104,6 +104,7 @@ public void replay(final TypedRecord event) {
   @Override
   public ProcessingResult process(
       final TypedRecord record, final ProcessingResultBuilder processingResultBuilder) {
+
     try (final var scope = new ProcessingResultBuilderScope(processingResultBuilder)) {
       TypedRecordProcessor<?> currentProcessor = null;
 
@@ -124,12 +125,7 @@ public ProcessingResult process(
 
       final boolean isNotOnBlacklist = !zeebeState.getBlackListState().isOnBlacklist(typedCommand);
       if (isNotOnBlacklist) {
-        currentProcessor.processRecord(
-            record,
-            (sep) -> {
-              processingResultBuilder.resetPostCommitTasks();
-              processingResultBuilder.appendPostCommitTask(sep::flush);
-            });
+        currentProcessor.processRecord(record);
       }
     }
     return processingResultBuilder.build();

engine/src/main/java/io/camunda/zeebe/engine/processing/EngineProcessors.java

@@ -24,7 +24,6 @@
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessor;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessorContext;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessors;
-import io.camunda.zeebe.engine.processing.streamprocessor.sideeffect.SideEffectQueue;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.Writers;
 import io.camunda.zeebe.engine.processing.timer.DueDateTimerChecker;
 import io.camunda.zeebe.engine.state.KeyGenerator;
@@ -67,7 +66,6 @@ public static TypedRecordProcessors createEngineProcessors(
 
     final var jobMetrics = new JobMetrics(partitionId);
     final var processEngineMetrics = new ProcessEngineMetrics(zeebeState.getPartitionId());
-    final var sideEffectQueue = new SideEffectQueue();
 
     final BpmnBehaviorsImpl bpmnBehaviors =
         createBehaviors(
@@ -77,8 +75,7 @@ public static TypedRecordProcessors createEngineProcessors(
             partitionsCount,
             timerChecker,
             jobMetrics,
-            processEngineMetrics,
-            sideEffectQueue);
+            processEngineMetrics);
 
     addDeploymentRelatedProcessorAndServices(
         bpmnBehaviors,
@@ -98,8 +95,7 @@ public static TypedRecordProcessors createEngineProcessors(
             typedRecordProcessors,
             subscriptionCommandSender,
             writers,
-            timerChecker,
-            sideEffectQueue);
+            timerChecker);
 
     JobEventProcessors.addJobProcessors(
         typedRecordProcessors,
@@ -121,10 +117,8 @@ private static BpmnBehaviorsImpl createBehaviors(
       final int partitionsCount,
       final DueDateTimerChecker timerChecker,
       final JobMetrics jobMetrics,
-      final ProcessEngineMetrics processEngineMetrics,
-      final SideEffectQueue sideEffectQueue) {
+      final ProcessEngineMetrics processEngineMetrics) {
     return new BpmnBehaviorsImpl(
-        sideEffectQueue,
         zeebeState,
         writers,
         jobMetrics,
@@ -140,16 +134,14 @@ private static TypedRecordProcessor<ProcessInstanceRecord> addProcessProcessors(
       final TypedRecordProcessors typedRecordProcessors,
       final SubscriptionCommandSender subscriptionCommandSender,
       final Writers writers,
-      final DueDateTimerChecker timerChecker,
-      final SideEffectQueue sideEffectQueue) {
+      final DueDateTimerChecker timerChecker) {
     return ProcessEventProcessors.addProcessProcessors(
         zeebeState,
         bpmnBehaviors,
         typedRecordProcessors,
         subscriptionCommandSender,
         timerChecker,
-        writers,
-        sideEffectQueue);
+        writers);
   }
 
   private static void addDeploymentRelatedProcessorAndServices(

engine/src/main/java/io/camunda/zeebe/engine/processing/ProcessEventProcessors.java

@@ -21,7 +21,6 @@
 import io.camunda.zeebe.engine.processing.processinstance.ProcessInstanceModificationProcessor;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessor;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessors;
-import io.camunda.zeebe.engine.processing.streamprocessor.sideeffect.SideEffectQueue;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.Writers;
 import io.camunda.zeebe.engine.processing.timer.CancelTimerProcessor;
 import io.camunda.zeebe.engine.processing.timer.DueDateTimerChecker;
@@ -51,8 +50,7 @@ public static TypedRecordProcessor<ProcessInstanceRecord> addProcessProcessors(
       final TypedRecordProcessors typedRecordProcessors,
       final SubscriptionCommandSender subscriptionCommandSender,
       final DueDateTimerChecker timerChecker,
-      final Writers writers,
-      final SideEffectQueue sideEffectQueue) {
+      final Writers writers) {
     final MutableProcessMessageSubscriptionState subscriptionState =
         zeebeState.getProcessMessageSubscriptionState();
     final var keyGenerator = zeebeState.getKeyGenerator();
@@ -63,8 +61,7 @@ public static TypedRecordProcessor<ProcessInstanceRecord> addProcessProcessors(
         writers, typedRecordProcessors, zeebeState.getElementInstanceState());
 
     final var bpmnStreamProcessor =
-        new BpmnStreamProcessor(
-            bpmnBehaviors, zeebeState, writers, sideEffectQueue, processEngineMetrics);
+        new BpmnStreamProcessor(bpmnBehaviors, zeebeState, writers, processEngineMetrics);
     addBpmnStepProcessor(typedRecordProcessors, bpmnStreamProcessor);
 
     addMessageStreamProcessors(

engine/src/main/java/io/camunda/zeebe/engine/processing/bpmn/BpmnStreamProcessor.java

@@ -15,8 +15,6 @@
 import io.camunda.zeebe.engine.processing.bpmn.behavior.BpmnStateTransitionBehavior;
 import io.camunda.zeebe.engine.processing.deployment.model.element.ExecutableFlowElement;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessor;
-import io.camunda.zeebe.engine.processing.streamprocessor.sideeffect.SideEffectProducer;
-import io.camunda.zeebe.engine.processing.streamprocessor.sideeffect.SideEffectQueue;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.TypedRejectionWriter;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.Writers;
 import io.camunda.zeebe.engine.state.immutable.ProcessState;
@@ -25,7 +23,6 @@
 import io.camunda.zeebe.protocol.record.RejectionType;
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceIntent;
 import io.camunda.zeebe.protocol.record.value.BpmnElementType;
-import java.util.function.Consumer;
 import org.slf4j.Logger;
 
 public final class BpmnStreamProcessor implements TypedRecordProcessor<ProcessInstanceRecord> {
@@ -34,7 +31,6 @@ public final class BpmnStreamProcessor implements TypedRecordProcessor<ProcessIn
 
   private final BpmnElementContextImpl context = new BpmnElementContextImpl();
 
-  private final SideEffectQueue sideEffectQueue;
   private final ProcessState processState;
   private final BpmnElementProcessors processors;
   private final ProcessInstanceStateTransitionGuard stateTransitionGuard;
@@ -46,7 +42,6 @@ public BpmnStreamProcessor(
       final BpmnBehaviors bpmnBehaviors,
       final MutableZeebeState zeebeState,
       final Writers writers,
-      final SideEffectQueue sideEffectQueue,
       final ProcessEngineMetrics processEngineMetrics) {
     processState = zeebeState.getProcessState();
 
@@ -61,7 +56,6 @@ public BpmnStreamProcessor(
             this::getContainerProcessor,
             writers);
     processors = new BpmnElementProcessors(bpmnBehaviors, stateTransitionBehavior);
-    this.sideEffectQueue = sideEffectQueue;
   }
 
   private BpmnElementContainerProcessor<ExecutableFlowElement> getContainerProcessor(
@@ -70,14 +64,9 @@ private BpmnElementContainerProcessor<ExecutableFlowElement> getContainerProcess
   }
 
   @Override
-  public void processRecord(
-      final TypedRecord<ProcessInstanceRecord> record,
-      final Consumer<SideEffectProducer> sideEffect) {
+  public void processRecord(final TypedRecord<ProcessInstanceRecord> record) {
 
     // initialize
-    sideEffectQueue.clear();
-    sideEffect.accept(sideEffectQueue);
-
     final var intent = (ProcessInstanceIntent) record.getIntent();
     final var recordValue = record.getValue();
 

engine/src/main/java/io/camunda/zeebe/engine/processing/bpmn/behavior/BpmnBehaviorsImpl.java

@@ -17,7 +17,6 @@
 import io.camunda.zeebe.engine.processing.common.EventTriggerBehavior;
 import io.camunda.zeebe.engine.processing.common.ExpressionProcessor;
 import io.camunda.zeebe.engine.processing.message.command.SubscriptionCommandSender;
-import io.camunda.zeebe.engine.processing.streamprocessor.sideeffect.SideEffects;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.Writers;
 import io.camunda.zeebe.engine.processing.timer.DueDateTimerChecker;
 import io.camunda.zeebe.engine.processing.variable.VariableBehavior;
@@ -45,7 +44,6 @@ public final class BpmnBehaviorsImpl implements BpmnBehaviors {
   private final ElementActivationBehavior elementActivationBehavior;
 
   public BpmnBehaviorsImpl(
-      final SideEffects sideEffects,
       final MutableZeebeState zeebeState,
       final Writers writers,
       final JobMetrics jobMetrics,
@@ -69,6 +67,7 @@ public BpmnBehaviorsImpl(
             expressionBehavior,
             subscriptionCommandSender,
             writers.state(),
+            writers.sideEffect(),
             timerChecker,
             partitionsCount);
 
@@ -94,8 +93,7 @@ public BpmnBehaviorsImpl(
         new BpmnVariableMappingBehavior(expressionBehavior, zeebeState, variableBehavior);
 
     eventSubscriptionBehavior =
-        new BpmnEventSubscriptionBehavior(
-            catchEventBehavior, eventTriggerBehavior, sideEffects, zeebeState);
+        new BpmnEventSubscriptionBehavior(catchEventBehavior, eventTriggerBehavior, zeebeState);
 
     incidentBehavior =
         new BpmnIncidentBehavior(zeebeState, zeebeState.getKeyGenerator(), writers.state());

engine/src/main/java/io/camunda/zeebe/engine/processing/bpmn/behavior/BpmnEventSubscriptionBehavior.java

@@ -13,7 +13,6 @@
 import io.camunda.zeebe.engine.processing.common.Failure;
 import io.camunda.zeebe.engine.processing.deployment.model.element.ExecutableCatchEventSupplier;
 import io.camunda.zeebe.engine.processing.deployment.model.element.ExecutableFlowElement;
-import io.camunda.zeebe.engine.processing.streamprocessor.sideeffect.SideEffects;
 import io.camunda.zeebe.engine.state.immutable.EventScopeInstanceState;
 import io.camunda.zeebe.engine.state.immutable.ProcessState;
 import io.camunda.zeebe.engine.state.immutable.ZeebeState;
@@ -27,19 +26,15 @@ public final class BpmnEventSubscriptionBehavior {
   private final EventScopeInstanceState eventScopeInstanceState;
   private final CatchEventBehavior catchEventBehavior;
 
-  private final SideEffects sideEffects;
-
   private final ProcessState processState;
   private final EventTriggerBehavior eventTriggerBehavior;
 
   public BpmnEventSubscriptionBehavior(
       final CatchEventBehavior catchEventBehavior,
       final EventTriggerBehavior eventTriggerBehavior,
-      final SideEffects sideEffects,
       final ZeebeState zeebeState) {
     this.catchEventBehavior = catchEventBehavior;
     this.eventTriggerBehavior = eventTriggerBehavior;
-    this.sideEffects = sideEffects;
 
     processState = zeebeState.getProcessState();
     eventScopeInstanceState = zeebeState.getEventScopeInstanceState();
@@ -50,11 +45,11 @@ public BpmnEventSubscriptionBehavior(
    */
   public <T extends ExecutableCatchEventSupplier> Either<Failure, Void> subscribeToEvents(
       final T element, final BpmnElementContext context) {
-    return catchEventBehavior.subscribeToEvents(context, element, sideEffects);
+    return catchEventBehavior.subscribeToEvents(context, element);
   }
 
   public void unsubscribeFromEvents(final BpmnElementContext context) {
-    catchEventBehavior.unsubscribeFromEvents(context.getElementInstanceKey(), sideEffects);
+    catchEventBehavior.unsubscribeFromEvents(context.getElementInstanceKey());
   }
 
   /**