fix: create new readers and writer for every async request

broker/src/main/java/io/camunda/zeebe/broker/transport/ApiRequestHandler.java

@@ -11,17 +11,33 @@
 import io.camunda.zeebe.broker.transport.AsyncApiRequestHandler.ResponseWriter;
 import io.camunda.zeebe.scheduler.future.ActorFuture;
 import io.camunda.zeebe.scheduler.future.CompletableActorFuture;
+import io.camunda.zeebe.transport.RequestHandler;
 import io.camunda.zeebe.util.Either;
+import java.util.function.Supplier;
 
 /**
- * @param <R>
- * @param <W>
+ * A {@link RequestHandler} that automatically decodes requests and encodes successful and error
+ * responses. Handling requests is synchronous, use {@link AsyncApiRequestHandler} if handling
+ * should be asynchronous.
+ *
+ * @param <R> a {@link RequestReader} that reads the request. Reset on every request.
+ * @param <W> a {@link ResponseWriter} that writes the response. Reset on every request.
  */
 public abstract class ApiRequestHandler<R extends RequestReader<?>, W extends ResponseWriter>
     extends AsyncApiRequestHandler<R, W> {
 
   protected ApiRequestHandler(final R requestReader, final W responseWriter) {
-    super(requestReader, responseWriter);
+    super(
+        () -> requestReader,
+        () -> responseWriter,
+        new Supplier<>() {
+          private final ErrorResponseWriter errorResponseWriter = new ErrorResponseWriter();
+
+          @Override
+          public ErrorResponseWriter get() {
+            return errorResponseWriter;
+          }
+        });
   }
 
   protected abstract Either<ErrorResponseWriter, W> handle(

broker/src/main/java/io/camunda/zeebe/broker/transport/AsyncApiRequestHandler.java

@@ -17,6 +17,7 @@
 import io.camunda.zeebe.util.Either;
 import io.camunda.zeebe.util.buffer.BufferReader;
 import io.camunda.zeebe.util.buffer.BufferWriter;
+import java.util.function.Supplier;
 import org.agrona.DirectBuffer;
 import org.agrona.sbe.MessageDecoderFlyweight;
 import org.slf4j.Logger;
@@ -33,13 +34,23 @@ public abstract class AsyncApiRequestHandler<R extends RequestReader<?>, W exten
     extends Actor implements RequestHandler {
 
   public static final Logger LOG = Loggers.TRANSPORT_LOGGER;
-  private final ErrorResponseWriter errorResponseWriter = new ErrorResponseWriter();
-  private final R requestReader;
-  private final W responseWriter;
+  private final Supplier<R> requestReaderSupplier;
+  private final Supplier<W> responseWriterSupplier;
 
-  protected AsyncApiRequestHandler(final R requestReader, final W responseWriter) {
-    this.requestReader = requestReader;
-    this.responseWriter = responseWriter;
+  private final Supplier<ErrorResponseWriter> errorResponseWriterSupplier;
+
+  protected AsyncApiRequestHandler(
+      final Supplier<R> requestReaderSupplier, final Supplier<W> responseWriterSupplier) {
+    this(requestReaderSupplier, responseWriterSupplier, ErrorResponseWriter::new);
+  }
+
+  protected AsyncApiRequestHandler(
+      final Supplier<R> requestReaderSupplier,
+      final Supplier<W> responseWriterSupplier,
+      final Supplier<ErrorResponseWriter> errorResponseWriterSupplier) {
+    this.requestReaderSupplier = requestReaderSupplier;
+    this.responseWriterSupplier = responseWriterSupplier;
+    this.errorResponseWriterSupplier = errorResponseWriterSupplier;
   }
 
   /**
@@ -79,6 +90,10 @@ private void handleRequest(
       final DirectBuffer buffer,
       final int offset,
       final int length) {
+    final var requestReader = requestReaderSupplier.get();
+    final var responseWriter = responseWriterSupplier.get();
+    final var errorResponseWriter = errorResponseWriterSupplier.get();
+
     requestReader.reset();
     responseWriter.reset();
     errorResponseWriter.reset();

broker/src/main/java/io/camunda/zeebe/broker/transport/adminapi/AdminApiRequestHandler.java

@@ -29,7 +29,7 @@ public class AdminApiRequestHandler
 
   public AdminApiRequestHandler(
       final AtomixServerTransport transport, final PartitionManagerImpl partitionManager) {
-    super(new ApiRequestReader(), new ApiResponseWriter());
+    super(ApiRequestReader::new, ApiResponseWriter::new);
     this.transport = transport;
     this.partitionManager = partitionManager;
     adminAccess = partitionManager.createAdminAccess(this);

broker/src/test/java/io/camunda/zeebe/broker/transport/ApiRequestHandlerTest.java

@@ -13,7 +13,6 @@
 
 import io.camunda.zeebe.broker.transport.AsyncApiRequestHandler.RequestReader;
 import io.camunda.zeebe.broker.transport.AsyncApiRequestHandler.ResponseWriter;
-import io.camunda.zeebe.scheduler.ActorControl;
 import io.camunda.zeebe.scheduler.testing.ControlledActorSchedulerRule;
 import io.camunda.zeebe.transport.ServerOutput;
 import io.camunda.zeebe.util.Either;
@@ -93,10 +92,6 @@ private static class TestApiRequestHandler
       super(requestReader, responseWriter);
     }
 
-    public ActorControl actor() {
-      return actor;
-    }
-
     @Override
     protected Either<ErrorResponseWriter, ResponseWriter> handle(
         final int partitionId,