refactor: remove stream writer proxy

The proxy was introduced to disable writing during reprocressing, we no longer do reprocessing so this is no longer necessary an can be removed.

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

@@ -66,8 +66,7 @@ void setup() {
 
   private RecordProcessorContextImpl createContext(
       final ProcessingScheduleService executor, final ZeebeDb zeebeDb) {
-    return new RecordProcessorContextImpl(
-        1, executor, zeebeDb, zeebeDb.createContext(), null, null, null);
+    return new RecordProcessorContextImpl(1, executor, zeebeDb, zeebeDb.createContext(), null);
   }
 
   @AfterEach

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

@@ -18,8 +18,6 @@
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessorContextImpl;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessorFactory;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessors;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedResponseWriter;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedStreamWriter;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.Writers;
 import io.camunda.zeebe.engine.state.EventApplier;
 import io.camunda.zeebe.engine.state.ZeebeDbState;
@@ -43,8 +41,6 @@ public class Engine implements RecordProcessor {
   private EventApplier eventApplier;
   private RecordProcessorMap recordProcessorMap;
   private ZeebeDbState zeebeState;
-  private LegacyTypedStreamWriter streamWriter;
-  private LegacyTypedResponseWriter responseWriter;
 
   private final ErrorRecord errorRecord = new ErrorRecord();
 
@@ -62,9 +58,6 @@ public Engine(final TypedRecordProcessorFactory typedRecordProcessorFactory) {
 
   @Override
   public void init(final RecordProcessorContext recordProcessorContext) {
-    streamWriter = recordProcessorContext.getStreamWriterProxy();
-    responseWriter = recordProcessorContext.getTypedResponseWriter();
-
     zeebeState =
         new ZeebeDbState(
             recordProcessorContext.getPartitionId(),

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

@@ -10,8 +10,6 @@
 import io.camunda.zeebe.db.TransactionContext;
 import io.camunda.zeebe.db.ZeebeDb;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessorListener;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedResponseWriter;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedStreamWriter;
 import io.camunda.zeebe.engine.state.EventApplier;
 import io.camunda.zeebe.engine.state.mutable.MutableZeebeState;
 import java.util.List;
@@ -27,10 +25,6 @@ public interface RecordProcessorContext {
 
   TransactionContext getTransactionContext();
 
-  LegacyTypedStreamWriter getStreamWriterProxy();
-
-  LegacyTypedResponseWriter getTypedResponseWriter();
-
   Function<MutableZeebeState, EventApplier> getEventApplierFactory();
 
   List<StreamProcessorLifecycleAware> getLifecycleListeners();

engine/src/main/java/io/camunda/zeebe/streamprocessor/RecordProcessorContextImpl.java

@@ -13,8 +13,6 @@
 import io.camunda.zeebe.engine.api.RecordProcessorContext;
 import io.camunda.zeebe.engine.api.StreamProcessorLifecycleAware;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessorListener;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedResponseWriter;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedStreamWriter;
 import io.camunda.zeebe.engine.state.EventApplier;
 import io.camunda.zeebe.engine.state.mutable.MutableZeebeState;
 import java.util.ArrayList;
@@ -27,8 +25,6 @@ public final class RecordProcessorContextImpl implements RecordProcessorContext
   private final ProcessingScheduleService scheduleService;
   private final ZeebeDb zeebeDb;
   private final TransactionContext transactionContext;
-  private final LegacyTypedStreamWriter streamWriter;
-  private final LegacyTypedResponseWriter responseWriter;
   private final Function<MutableZeebeState, EventApplier> eventApplierFactory;
   private final List<StreamProcessorLifecycleAware> lifecycleListeners = new ArrayList<>();
   private StreamProcessorListener streamProcessorListener;
@@ -38,15 +34,11 @@ public RecordProcessorContextImpl(
       final ProcessingScheduleService scheduleService,
       final ZeebeDb zeebeDb,
       final TransactionContext transactionContext,
-      final LegacyTypedStreamWriter streamWriter,
-      final LegacyTypedResponseWriter responseWriter,
       final Function<MutableZeebeState, EventApplier> eventApplierFactory) {
     this.partitionId = partitionId;
     this.scheduleService = scheduleService;
     this.zeebeDb = zeebeDb;
     this.transactionContext = transactionContext;
-    this.streamWriter = streamWriter;
-    this.responseWriter = responseWriter;
     this.eventApplierFactory = eventApplierFactory;
   }
 
@@ -70,16 +62,6 @@ public TransactionContext getTransactionContext() {
     return transactionContext;
   }
 
-  @Override
-  public LegacyTypedStreamWriter getStreamWriterProxy() {
-    return streamWriter;
-  }
-
-  @Override
-  public LegacyTypedResponseWriter getTypedResponseWriter() {
-    return responseWriter;
-  }
-
   @Override
   public Function<MutableZeebeState, EventApplier> getEventApplierFactory() {
     return eventApplierFactory;

engine/src/main/java/io/camunda/zeebe/streamprocessor/StreamProcessor.java

@@ -176,8 +176,6 @@ protected void onActorStarted() {
 
       replayStateMachine =
           new ReplayStateMachine(recordProcessor, streamProcessorContext, this::shouldProcessNext);
-      // disable writing to the log stream
-      streamProcessorContext.disableLogStreamWriter();
 
       openFuture.complete(null);
       replayCompletedFuture = replayStateMachine.startRecover(snapshotPosition);
@@ -281,9 +279,6 @@ private void onRetrievingWriter(
 
       phase = Phase.PROCESSING;
 
-      // enable writing records to the stream
-      streamProcessorContext.enableLogStreamWriter();
-
       processingStateMachine =
           new ProcessingStateMachine(
               streamProcessorContext, this::shouldProcessNext, recordProcessor);
@@ -328,8 +323,6 @@ private void initEngine() {
             streamProcessorContext.getScheduleService(),
             zeebeDb,
             streamProcessorContext.getTransactionContext(),
-            streamProcessorContext.getLogStreamWriter(),
-            streamProcessorContext.getTypedResponseWriter(),
             eventApplierFactory);
     recordProcessor.init(engineContext);
 

engine/src/main/java/io/camunda/zeebe/streamprocessor/StreamProcessorContext.java

@@ -10,13 +10,11 @@
 import io.camunda.zeebe.db.TransactionContext;
 import io.camunda.zeebe.engine.api.ProcessingScheduleService;
 import io.camunda.zeebe.engine.api.ReadonlyStreamProcessorContext;
-import io.camunda.zeebe.engine.processing.bpmn.behavior.LegacyTypedStreamWriterProxy;
 import io.camunda.zeebe.engine.processing.streamprocessor.RecordValues;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessorListener;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedResponseWriterImpl;
 import io.camunda.zeebe.engine.processing.streamprocessor.writers.LegacyTypedStreamWriter;
-import io.camunda.zeebe.engine.processing.streamprocessor.writers.NoopLegacyTypedStreamWriter;
 import io.camunda.zeebe.engine.state.EventApplier;
 import io.camunda.zeebe.engine.state.KeyGeneratorControls;
 import io.camunda.zeebe.engine.state.ZeebeDbState;
@@ -32,14 +30,10 @@ public final class StreamProcessorContext implements ReadonlyStreamProcessorCont
 
   private static final StreamProcessorListener NOOP_LISTENER = processedCommand -> {};
 
-  private final LegacyTypedStreamWriterProxy streamWriterProxy = new LegacyTypedStreamWriterProxy();
-  private final NoopLegacyTypedStreamWriter noopTypedStreamWriter =
-      new NoopLegacyTypedStreamWriter();
-
   private ActorControl actor;
   private LogStream logStream;
   private LogStreamReader logStreamReader;
-  private LegacyTypedStreamWriter logStreamWriter = noopTypedStreamWriter;
+  private LegacyTypedStreamWriter logStreamWriter;
   private LegacyTypedResponseWriterImpl typedResponseWriter;
 
   private RecordValues recordValues;
@@ -59,7 +53,7 @@ public final class StreamProcessorContext implements ReadonlyStreamProcessorCont
 
   public StreamProcessorContext actor(final ActorControl actor) {
     this.actor = actor;
-    processingScheduleService = new ProcessingScheduleServiceImpl(actor, streamWriterProxy);
+    processingScheduleService = new ProcessingScheduleServiceImpl(actor);
     return this;
   }
 
@@ -75,7 +69,7 @@ public LogStream getLogStream() {
 
   @Override
   public LegacyTypedStreamWriter getLogStreamWriter() {
-    return streamWriterProxy;
+    return logStreamWriter;
   }
 
   @Override
@@ -188,14 +182,6 @@ public StreamProcessorListener getStreamProcessorListener() {
     return streamProcessorListener;
   }
 
-  public void enableLogStreamWriter() {
-    streamWriterProxy.wrap(logStreamWriter);
-  }
-
-  public void disableLogStreamWriter() {
-    streamWriterProxy.wrap(noopTypedStreamWriter);
-  }
-
   public StreamProcessorMode getProcessorMode() {
     return streamProcessorMode;
   }

engine/src/test/java/io/camunda/zeebe/engine/processing/job/JobTimeoutTriggerTest.java

@@ -45,7 +45,6 @@ public void setUp() {
 
     final StreamProcessorContext streamProcessorContext =
         new StreamProcessorContext().actor(someActor).logStreamWriter(typedStreamWriter);
-    streamProcessorContext.enableLogStreamWriter();
     jobTimeoutTrigger.onRecovered(streamProcessorContext);
 
     IntStream.range(0, 3)