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.

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

@@ -16,8 +16,6 @@
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessor;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessorContextImpl;
 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;
@@ -41,8 +39,6 @@ public class Engine implements RecordProcessor<EngineContext> {
   private EventApplier eventApplier;
   private RecordProcessorMap recordProcessorMap;
   private ZeebeDbState zeebeState;
-  private LegacyTypedStreamWriter streamWriter;
-  private LegacyTypedResponseWriter responseWriter;
 
   private final ErrorRecord errorRecord = new ErrorRecord();
 
@@ -55,9 +51,6 @@ public Engine() {}
 
   @Override
   public void init(final EngineContext engineContext) {
-    streamWriter = engineContext.getStreamWriterProxy();
-    responseWriter = engineContext.getTypedResponseWriter();
-
     zeebeState =
         new ZeebeDbState(
             engineContext.getPartitionId(),

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

@@ -13,8 +13,6 @@
 import io.camunda.zeebe.engine.api.StreamProcessorLifecycleAware;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessorListener;
 import io.camunda.zeebe.engine.processing.streamprocessor.TypedRecordProcessorFactory;
-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.Collections;
@@ -27,8 +25,6 @@ public final class EngineContext {
   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 TypedRecordProcessorFactory typedRecordProcessorFactory;
   private List<StreamProcessorLifecycleAware> lifecycleListeners = Collections.EMPTY_LIST;
@@ -39,16 +35,12 @@ public EngineContext(
       final ProcessingScheduleService scheduleService,
       final ZeebeDb zeebeDb,
       final TransactionContext transactionContext,
-      final LegacyTypedStreamWriter streamWriter,
-      final LegacyTypedResponseWriter responseWriter,
       final Function<MutableZeebeState, EventApplier> eventApplierFactory,
       final TypedRecordProcessorFactory typedRecordProcessorFactory) {
     this.partitionId = partitionId;
     this.scheduleService = scheduleService;
     this.zeebeDb = zeebeDb;
     this.transactionContext = transactionContext;
-    this.streamWriter = streamWriter;
-    this.responseWriter = responseWriter;
     this.eventApplierFactory = eventApplierFactory;
     this.typedRecordProcessorFactory = typedRecordProcessorFactory;
   }
@@ -69,14 +61,6 @@ public TransactionContext getTransactionContext() {
     return transactionContext;
   }
 
-  public LegacyTypedStreamWriter getStreamWriterProxy() {
-    return streamWriter;
-  }
-
-  public LegacyTypedResponseWriter getTypedResponseWriter() {
-    return responseWriter;
-  }
-
   public Function<MutableZeebeState, EventApplier> getEventApplierFactory() {
     return eventApplierFactory;
   }

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

@@ -182,8 +182,6 @@ protected void onActorStarted() {
 
       replayStateMachine =
           new ReplayStateMachine(engine, streamProcessorContext, this::shouldProcessNext);
-      // disable writing to the log stream
-      streamProcessorContext.disableLogStreamWriter();
 
       openFuture.complete(null);
       replayCompletedFuture = replayStateMachine.startRecover(snapshotPosition);
@@ -287,9 +285,6 @@ private void onRetrievingWriter(
 
       phase = Phase.PROCESSING;
 
-      // enable writing records to the stream
-      streamProcessorContext.enableLogStreamWriter();
-
       processingStateMachine =
           new ProcessingStateMachine(streamProcessorContext, this::shouldProcessNext, engine);
 
@@ -333,8 +328,6 @@ private void initEngine() {
             streamProcessorContext.getScheduleService(),
             zeebeDb,
             streamProcessorContext.getTransactionContext(),
-            streamProcessorContext.getLogStreamWriter(),
-            streamProcessorContext.getTypedResponseWriter(),
             eventApplierFactory,
             typedRecordProcessorFactory);
     engine.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)