merge: #10076

10076: Engine abstraction remove legacy writer r=Zelldon a=pihme

## Description

- Introduces a new task that can be scheduled by the engine and that can return records to be written
- Replaces all references to `LegacyTypedStreamWriter`outside of stream processor with scheduling of the new tasks
- The overall result is that the engine no longer depends on any writer that writes directly to the stream. Stream processor is now in full control of when and how records shall be written (Heureka!)

## Related issues

related to #9724, #9730, #9725



Co-authored-by: pihme <pihme@users.noreply.github.com>
Co-authored-by: Christopher Zell <zelldon91@googlemail.com>

backup/src/test/java/io/camunda/zeebe/backup/processing/MockProcessingResult.java

@@ -7,10 +7,10 @@
  */
 package io.camunda.zeebe.backup.processing;
 
+import io.camunda.zeebe.engine.api.CommandResponseWriter;
 import io.camunda.zeebe.engine.api.PostCommitTask;
 import io.camunda.zeebe.engine.api.ProcessingResult;
 import io.camunda.zeebe.engine.api.ProcessingResultBuilder;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import io.camunda.zeebe.logstreams.log.LogStreamBatchWriter;
 import io.camunda.zeebe.msgpack.UnpackedObject;
 import io.camunda.zeebe.protocol.record.RecordType;

broker/src/main/java/io/camunda/zeebe/broker/system/partitions/PartitionStartupAndTransitionContextImpl.java

@@ -25,9 +25,9 @@
 import io.camunda.zeebe.broker.transport.partitionapi.InterPartitionCommandReceiverActor;
 import io.camunda.zeebe.broker.transport.partitionapi.InterPartitionCommandSenderService;
 import io.camunda.zeebe.db.ZeebeDb;
+import io.camunda.zeebe.engine.api.CommandResponseWriter;
 import io.camunda.zeebe.engine.api.TypedRecord;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessorFactory;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import io.camunda.zeebe.engine.state.QueryService;
 import io.camunda.zeebe.logstreams.log.LogStream;
 import io.camunda.zeebe.scheduler.ActorControl;

broker/src/main/java/io/camunda/zeebe/broker/system/partitions/PartitionTransitionContext.java

@@ -20,9 +20,9 @@
 import io.camunda.zeebe.broker.transport.partitionapi.InterPartitionCommandReceiverActor;
 import io.camunda.zeebe.broker.transport.partitionapi.InterPartitionCommandSenderService;
 import io.camunda.zeebe.db.ZeebeDb;
+import io.camunda.zeebe.engine.api.CommandResponseWriter;
 import io.camunda.zeebe.engine.api.TypedRecord;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessorFactory;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import io.camunda.zeebe.engine.state.QueryService;
 import io.camunda.zeebe.logstreams.log.LogStream;
 import io.camunda.zeebe.scheduler.ActorSchedulingService;

broker/src/main/java/io/camunda/zeebe/broker/transport/commandapi/CommandApiService.java

@@ -7,8 +7,8 @@
  */
 package io.camunda.zeebe.broker.transport.commandapi;
 
+import io.camunda.zeebe.engine.api.CommandResponseWriter;
 import io.camunda.zeebe.engine.api.TypedRecord;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import java.util.function.Consumer;
 
 public interface CommandApiService {

broker/src/main/java/io/camunda/zeebe/broker/transport/commandapi/CommandApiServiceImpl.java

@@ -14,8 +14,8 @@
 import io.camunda.zeebe.broker.transport.backpressure.PartitionAwareRequestLimiter;
 import io.camunda.zeebe.broker.transport.backpressure.RequestLimiter;
 import io.camunda.zeebe.broker.transport.queryapi.QueryApiRequestHandler;
+import io.camunda.zeebe.engine.api.CommandResponseWriter;
 import io.camunda.zeebe.engine.api.TypedRecord;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import io.camunda.zeebe.engine.state.QueryService;
 import io.camunda.zeebe.logstreams.log.LogStream;
 import io.camunda.zeebe.protocol.impl.encoding.BrokerInfo;

broker/src/main/java/io/camunda/zeebe/broker/transport/commandapi/CommandResponseWriterImpl.java

@@ -11,7 +11,7 @@
 import static io.camunda.zeebe.protocol.record.ExecuteCommandResponseEncoder.partitionIdNullValue;
 import static io.camunda.zeebe.protocol.record.ExecuteCommandResponseEncoder.valueHeaderLength;
 
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
+import io.camunda.zeebe.engine.api.CommandResponseWriter;
 import io.camunda.zeebe.protocol.Protocol;
 import io.camunda.zeebe.protocol.record.ExecuteCommandResponseEncoder;
 import io.camunda.zeebe.protocol.record.MessageHeaderEncoder;

broker/src/test/java/io/camunda/zeebe/broker/system/partitions/TestPartitionTransitionContext.java

@@ -23,9 +23,9 @@
 import io.camunda.zeebe.broker.transport.partitionapi.InterPartitionCommandReceiverActor;
 import io.camunda.zeebe.broker.transport.partitionapi.InterPartitionCommandSenderService;
 import io.camunda.zeebe.db.ZeebeDb;
+import io.camunda.zeebe.engine.api.CommandResponseWriter;
 import io.camunda.zeebe.engine.api.TypedRecord;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessorFactory;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import io.camunda.zeebe.engine.state.QueryService;
 import io.camunda.zeebe.logstreams.log.LogStream;
 import io.camunda.zeebe.scheduler.ActorSchedulingService;

engine/src/main/java/io/camunda/zeebe/engine/processing/streamprocessor/writers/CommandResponseWriter.java → engine/src/main/java/io/camunda/zeebe/engine/api/CommandResponseWriter.java

@@ -5,7 +5,7 @@
  * Licensed under the Zeebe Community License 1.1. You may not use this file
  * except in compliance with the Zeebe Community License 1.1.
  */
-package io.camunda.zeebe.engine.processing.streamprocessor.writers;
+package io.camunda.zeebe.engine.api;
 
 import io.camunda.zeebe.protocol.record.RecordType;
 import io.camunda.zeebe.protocol.record.RejectionType;

engine/src/main/java/io/camunda/zeebe/engine/api/EmptyProcessingResult.java

@@ -7,7 +7,6 @@
  */
 package io.camunda.zeebe.engine.api;
 
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import io.camunda.zeebe.logstreams.log.LogStreamBatchWriter;
 
 public final class EmptyProcessingResult implements ProcessingResult {

engine/src/main/java/io/camunda/zeebe/engine/api/ProcessingResult.java

@@ -7,7 +7,6 @@
  */
 package io.camunda.zeebe.engine.api;
 
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import io.camunda.zeebe.logstreams.log.LogStreamBatchWriter;
 
 /**

engine/src/main/java/io/camunda/zeebe/engine/api/ProcessingScheduleService.java

@@ -13,7 +13,9 @@
 
 public interface ProcessingScheduleService {
 
-  void runDelayed(Duration delay, Runnable followUpTask);
+  void runDelayed(Duration delay, Runnable task);
+
+  void runDelayed(Duration delay, Task task);
 
   <T> void runOnCompletion(ActorFuture<T> precedingTask, BiConsumer<T, Throwable> followUpTask);
 
@@ -40,4 +42,6 @@ default void runAtFixedRate(final Duration delay, final Runnable task) {
           }
         });
   }
+
+  void runAtFixedRate(final Duration delay, final Task task);
 }

engine/src/main/java/io/camunda/zeebe/engine/api/ReadonlyStreamProcessorContext.java

@@ -7,7 +7,6 @@
  */
 package io.camunda.zeebe.engine.api;
 
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedStreamWriter;
 import io.camunda.zeebe.engine.state.mutable.MutableZeebeState;
 import io.camunda.zeebe.logstreams.log.LogStream;
 
@@ -18,13 +17,9 @@ public interface ReadonlyStreamProcessorContext {
   /**
    * @return the logstream, on which the processor runs
    */
+  @Deprecated // only used in EngineRule; TODO remove this
   LogStream getLogStream();
 
-  /**
-   * @return the actual log stream writer, used to write any record
-   */
-  LegacyTypedStreamWriter getLogStreamWriter();
-
   /**
    * @return the state, where the data is stored during processing
    */

engine/src/main/java/io/camunda/zeebe/engine/api/Task.java

@@ -0,0 +1,14 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.engine.api;
+
+/** Here the interface is just a suggestion. Can be whatever PDT team thinks is best to work with */
+public interface Task {
+
+  TaskResult execute(TaskResultBuilder taskResultBuilder);
+}

engine/src/main/java/io/camunda/zeebe/engine/api/TaskResult.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.engine.api;
+
+import io.camunda.zeebe.logstreams.log.LogStreamBatchWriter;
+
+/** Here the interface is just a suggestion. Can be whatever PDT team thinks is best to work with */
+public interface TaskResult {
+
+  long writeRecordsToStream(LogStreamBatchWriter logStreamBatchWriter);
+}

engine/src/main/java/io/camunda/zeebe/engine/api/TaskResultBuilder.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.engine.api;
+
+import io.camunda.zeebe.protocol.record.RecordValue;
+import io.camunda.zeebe.protocol.record.intent.Intent;
+
+/** Here the interface is just a suggestion. Can be whatever PDT team thinks is best to work with */
+public interface TaskResultBuilder {
+
+  /**
+   * Appends a record to the result
+   *
+   * @return returns itself for method chaining
+   */
+  TaskResultBuilder appendCommandRecord(
+      final long key, final Intent intent, final RecordValue value);
+
+  TaskResult build();
+}

engine/src/main/java/io/camunda/zeebe/engine/processing/job/JobBackoffChecker.java

@@ -25,15 +25,14 @@ public JobBackoffChecker(final JobState jobState) {
     backOffDueDateChecker =
         new DueDateChecker(
             BACKOFF_RESOLUTION,
-            typedCommandWriter ->
+            taskResultBuilder ->
                 jobState.findBackedOffJobs(
                     ActorClock.currentTimeMillis(),
                     (key, record) -> {
-                      typedCommandWriter.reset();
-                      typedCommandWriter.appendFollowUpCommand(
+                      taskResultBuilder.appendCommandRecord(
                           key, JobIntent.RECUR_AFTER_BACKOFF, record);
 
-                      return typedCommandWriter.flush() >= 0;
+                      return true;
                     }));
   }
 

engine/src/main/java/io/camunda/zeebe/engine/processing/job/JobTimeoutTrigger.java

@@ -11,7 +11,9 @@
 
 import io.camunda.zeebe.engine.api.ReadonlyStreamProcessorContext;
 import io.camunda.zeebe.engine.api.StreamProcessorLifecycleAware;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedStreamWriter;
+import io.camunda.zeebe.engine.api.Task;
+import io.camunda.zeebe.engine.api.TaskResult;
+import io.camunda.zeebe.engine.api.TaskResultBuilder;
 import io.camunda.zeebe.engine.state.immutable.JobState;
 import io.camunda.zeebe.protocol.record.intent.JobIntent;
 import java.time.Duration;
@@ -22,19 +24,20 @@ public final class JobTimeoutTrigger implements StreamProcessorLifecycleAware {
 
   private boolean shouldReschedule = false;
 
-  private LegacyTypedStreamWriter writer;
   private ReadonlyStreamProcessorContext processingContext;
+  private final Task deactivateTimedOutJobs;
 
   public JobTimeoutTrigger(final JobState state) {
     this.state = state;
+    deactivateTimedOutJobs = new DeactivateTimeOutJobs();
   }
 
   @Override
   public void onRecovered(final ReadonlyStreamProcessorContext processingContext) {
     this.processingContext = processingContext;
     shouldReschedule = true;
+
     scheduleDeactivateTimedOutJobsTask();
-    writer = processingContext.getLogStreamWriter();
   }
 
   @Override
@@ -62,25 +65,28 @@ public void onResumed() {
   private void scheduleDeactivateTimedOutJobsTask() {
     processingContext
         .getScheduleService()
-        .runDelayed(TIME_OUT_POLLING_INTERVAL, this::deactivateTimedOutJobs);
+        .runDelayed(TIME_OUT_POLLING_INTERVAL, deactivateTimedOutJobs);
   }
 
   private void cancelTimer() {
     shouldReschedule = false;
   }
 
-  void deactivateTimedOutJobs() {
-    final long now = currentTimeMillis();
-    state.forEachTimedOutEntry(
-        now,
-        (key, record) -> {
-          writer.reset();
-          writer.appendFollowUpCommand(key, JobIntent.TIME_OUT, record);
+  private final class DeactivateTimeOutJobs implements Task {
 
-          return writer.flush() >= 0;
-        });
-    if (shouldReschedule) {
-      scheduleDeactivateTimedOutJobsTask();
+    @Override
+    public TaskResult execute(final TaskResultBuilder taskResultBuilder) {
+      final long now = currentTimeMillis();
+      state.forEachTimedOutEntry(
+          now,
+          (key, record) -> {
+            taskResultBuilder.appendCommandRecord(key, JobIntent.TIME_OUT, record);
+            return true;
+          });
+      if (shouldReschedule) {
+        scheduleDeactivateTimedOutJobsTask();
+      }
+      return taskResultBuilder.build();
     }
   }
 }

engine/src/main/java/io/camunda/zeebe/engine/processing/message/MessageObserver.java

@@ -38,8 +38,7 @@ public MessageObserver(
   public void onRecovered(final ReadonlyStreamProcessorContext context) {
     final var scheduleService = context.getScheduleService();
     // it is safe to reuse the write because we running in the same actor/thread
-    final MessageTimeToLiveChecker timeToLiveChecker =
-        new MessageTimeToLiveChecker(context.getLogStreamWriter(), messageState);
+    final MessageTimeToLiveChecker timeToLiveChecker = new MessageTimeToLiveChecker(messageState);
     scheduleService.runAtFixedRate(MESSAGE_TIME_TO_LIVE_CHECK_INTERVAL, timeToLiveChecker);
 
     final PendingMessageSubscriptionChecker pendingSubscriptionChecker =

engine/src/main/java/io/camunda/zeebe/engine/processing/message/MessageTimeToLiveChecker.java

@@ -7,33 +7,35 @@
  */
 package io.camunda.zeebe.engine.processing.message;
 
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedStreamWriter;
+import io.camunda.zeebe.engine.api.Task;
+import io.camunda.zeebe.engine.api.TaskResult;
+import io.camunda.zeebe.engine.api.TaskResultBuilder;
 import io.camunda.zeebe.engine.state.immutable.MessageState;
 import io.camunda.zeebe.engine.state.message.StoredMessage;
 import io.camunda.zeebe.protocol.impl.record.value.message.MessageRecord;
 import io.camunda.zeebe.protocol.record.intent.MessageIntent;
 import io.camunda.zeebe.scheduler.clock.ActorClock;
 
-public final class MessageTimeToLiveChecker implements Runnable {
+public final class MessageTimeToLiveChecker implements Task {
 
-  private final LegacyTypedStreamWriter writer;
   private final MessageState messageState;
 
   private final MessageRecord deleteMessageCommand = new MessageRecord();
 
-  public MessageTimeToLiveChecker(
-      final LegacyTypedStreamWriter writer, final MessageState messageState) {
-    this.writer = writer;
+  public MessageTimeToLiveChecker(final MessageState messageState) {
     this.messageState = messageState;
   }
 
   @Override
-  public void run() {
+  public TaskResult execute(final TaskResultBuilder taskResultBuilder) {
     messageState.visitMessagesWithDeadlineBefore(
-        ActorClock.currentTimeMillis(), this::writeDeleteMessageCommand);
+        ActorClock.currentTimeMillis(),
+        message -> writeDeleteMessageCommand(message, taskResultBuilder));
+    return taskResultBuilder.build();
   }
 
-  private boolean writeDeleteMessageCommand(final StoredMessage storedMessage) {
+  private boolean writeDeleteMessageCommand(
+      final StoredMessage storedMessage, final TaskResultBuilder taskResultBuilder) {
     final var message = storedMessage.getMessage();
 
     deleteMessageCommand.reset();
@@ -47,11 +49,8 @@ private boolean writeDeleteMessageCommand(final StoredMessage storedMessage) {
       deleteMessageCommand.setMessageId(message.getMessageIdBuffer());
     }
 
-    writer.reset();
-    writer.appendFollowUpCommand(
+    taskResultBuilder.appendCommandRecord(
         storedMessage.getMessageKey(), MessageIntent.EXPIRE, deleteMessageCommand);
-
-    final long position = writer.flush();
-    return position > 0;
+    return true;
   }
 }