allow concurrent Batch deserialization (#17050) (#17106)

Currently the deserialization is behind the readBatch's lock, so any large batch will take time deserializing, causing any other Queue writer (e.g. netty executor threads) and any other Queue reader (pipeline worker) to block.

This commit moves the deserialization out of the lock, allowing multiple pipeline workers to deserialize batches concurrently.

- add intermediate batch-holder from `Queue` methods
- make the intermediate batch-holder a private inner class of `Queue` with a descriptive name `SerializedBatchHolder`

Co-authored-by: Ry Biesemeyer <[email protected]>
(cherry picked from commit 637f447)

Co-authored-by: João Duarte <[email protected]>

Author: github-actions[bot]

logstash-core/src/main/java/org/logstash/ackedqueue/Queue.java

@@ -590,13 +590,18 @@ public void ensurePersistedUpto(long seqNum) throws IOException{
      * @throws IOException if an IO error occurs
      */
     public synchronized Batch nonBlockReadBatch(int limit) throws IOException {
+        final SerializedBatchHolder serializedBatchHolder;
         lock.lock();
         try {
             Page p = nextReadPage();
-            return (isHeadPage(p) && p.isFullyRead()) ? null : readPageBatch(p, limit, 0L);
+            if (isHeadPage(p) && p.isFullyRead()) {
+                return null;
+            }
+            serializedBatchHolder = readPageBatch(p, limit, 0L);
         } finally {
             lock.unlock();
         }
+        return serializedBatchHolder.deserialize();
     }
 
     /**
@@ -607,7 +612,11 @@ public synchronized Batch nonBlockReadBatch(int limit) throws IOException {
      * @throws QueueRuntimeException if queue is closed
      * @throws IOException if an IO error occurs
      */
-    public synchronized Batch readBatch(int limit, long timeout) throws IOException {
+    public Batch readBatch(int limit, long timeout) throws IOException {
+        return readSerializedBatch(limit, timeout).deserialize();
+    }
+
+    private synchronized SerializedBatchHolder readSerializedBatch(int limit, long timeout) throws IOException {
         lock.lock();
 
         try {
@@ -618,15 +627,15 @@ public synchronized Batch readBatch(int limit, long timeout) throws IOException
     }
 
     /**
-     * read a {@link Batch} from the given {@link Page}. If the page is a head page, try to maximize the
+     * read a {@link SerializedBatchHolder} from the given {@link Page}. If the page is a head page, try to maximize the
      * batch size by waiting for writes.
      * @param p the {@link Page} to read from.
      * @param limit size limit of the batch to read.
      * @param timeout  the maximum time to wait in milliseconds on write operations.
      * @return {@link Batch} with read elements or null if nothing was read
      * @throws IOException if an IO error occurs
      */
-    private Batch readPageBatch(Page p, int limit, long timeout) throws IOException {
+    private SerializedBatchHolder readPageBatch(Page p, int limit, long timeout) throws IOException {
         int left = limit;
         final List<byte[]> elements = new ArrayList<>(limit);
 
@@ -678,7 +687,7 @@ private Batch readPageBatch(Page p, int limit, long timeout) throws IOException
             removeUnreadPage(p);
         }
 
-        return new Batch(elements, firstSeqNum, this);
+        return new SerializedBatchHolder(elements, firstSeqNum);
     }
 
     /**
@@ -894,4 +903,18 @@ private static boolean containsSeq(final Page page, final long seqNum) {
         final long pMaxSeq = pMinSeq + (long) page.getElementCount();
         return seqNum >= pMinSeq && seqNum < pMaxSeq;
     }
+
+    class SerializedBatchHolder {
+        private final List<byte[]> elements;
+        private final long firstSeqNum;
+
+        private SerializedBatchHolder(List<byte[]> elements, long firstSeqNum) {
+            this.elements = elements;
+            this.firstSeqNum = firstSeqNum;
+        }
+
+        private Batch deserialize() {
+            return new Batch(elements, firstSeqNum, Queue.this);
+        }
+    }
 }