merge: #8277

8277: [Backport stable/1.2] Fix data race between `AtomixServerTransport` and `RequestHandler`s r=oleschoenburg a=github-actions[bot] # Description Backport of #8272 to `stable/1.2`. relates to #8269 #8257 Co-authored-by: Ole Schönburg <[email protected]>
camunda · Nov 25, 2021 · 5fd6471 · 5fd6471
2 parents 764962f + cf9d5a8
commit 5fd6471
Showing 1 changed file with 6 additions and 5 deletions.
diff --git a/transport/src/main/java/io/camunda/zeebe/transport/impl/AtomixServerTransport.java b/transport/src/main/java/io/camunda/zeebe/transport/impl/AtomixServerTransport.java
@@ -17,7 +17,6 @@
 import java.util.Arrays;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
-import org.agrona.DirectBuffer;
 import org.agrona.collections.Int2ObjectHashMap;
 import org.agrona.collections.Long2ObjectHashMap;
 import org.agrona.concurrent.UnsafeBuffer;
@@ -33,15 +32,13 @@ public class AtomixServerTransport extends Actor implements ServerTransport {
   private final Int2ObjectHashMap<Long2ObjectHashMap<CompletableFuture<byte[]>>>
       partitionsRequestMap;
   private final AtomicLong requestCount;
-  private final DirectBuffer reusableRequestBuffer;
   private final MessagingService messagingService;
   private final String actorName;
 
   public AtomixServerTransport(final int nodeId, final MessagingService messagingService) {
     this.messagingService = messagingService;
     partitionsRequestMap = new Int2ObjectHashMap<>();
     requestCount = new AtomicLong(0);
-    reusableRequestBuffer = new UnsafeBuffer(0, 0);
     actorName = buildActorName(nodeId, "ServerTransport");
   }
 
@@ -122,9 +119,13 @@ private CompletableFuture<byte[]> handleAtomixRequest(
           }
 
           try {
-            reusableRequestBuffer.wrap(requestBytes);
             requestHandler.onRequest(
-                this, partitionId, requestId, reusableRequestBuffer, 0, requestBytes.length);
+                this,
+                partitionId,
+                requestId,
+                new UnsafeBuffer(requestBytes),
+                0,
+                requestBytes.length);
             if (LOG.isTraceEnabled()) {
               LOG.trace(
                   "Handled request {} for topic {}",