Improve serialization of translog operations

server/src/main/java/org/elasticsearch/common/bytes/ReleasableBytesReference.java

@@ -51,6 +51,13 @@ public static ReleasableBytesReference wrap(BytesReference reference) {
         return reference.length() == 0 ? empty() : new ReleasableBytesReference(reference, ALWAYS_REFERENCED);
     }
 
+    public static BytesReference unwrap(BytesReference reference) {
+        if (reference instanceof ReleasableBytesReference releasable) {
+            return releasable.delegate;
+        }
+        return reference;
+    }
+
     @Override
     public void incRef() {
         refCounted.incRef();
@@ -278,6 +285,11 @@ public int arrayOffset() {
         return delegate.arrayOffset();
     }
 
+    public BytesReference delegate() {
+        assert hasReferences();
+        return delegate;
+    }
+
     private static final class RefCountedReleasable extends AbstractRefCounted {
 
         private final Releasable releasable;

server/src/main/java/org/elasticsearch/common/io/stream/RecyclerBytesStreamOutput.java

@@ -15,14 +15,12 @@
 import org.elasticsearch.common.bytes.CompositeBytesReference;
 import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.recycler.Recycler;
+import org.elasticsearch.common.util.ByteUtils;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.lang.invoke.VarHandle;
-import java.nio.ByteOrder;
 import java.util.ArrayList;
 import java.util.Objects;
 
@@ -33,11 +31,6 @@
  */
 public class RecyclerBytesStreamOutput extends BytesStream implements Releasable {
 
-    static final VarHandle VH_BE_INT = MethodHandles.byteArrayViewVarHandle(int[].class, ByteOrder.BIG_ENDIAN);
-    static final VarHandle VH_LE_INT = MethodHandles.byteArrayViewVarHandle(int[].class, ByteOrder.LITTLE_ENDIAN);
-    static final VarHandle VH_BE_LONG = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.BIG_ENDIAN);
-    static final VarHandle VH_LE_LONG = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.LITTLE_ENDIAN);
-
     private ArrayList<Recycler.V<BytesRef>> pages = new ArrayList<>(8);
     private final Recycler<BytesRef> recycler;
     private final int pageSize;
@@ -177,7 +170,7 @@ public void writeInt(int i) throws IOException {
             super.writeInt(i);
         } else {
             BytesRef currentPage = currentBytesRef;
-            VH_BE_INT.set(currentPage.bytes, currentPage.offset + currentPageOffset, i);
+            ByteUtils.writeIntBE(i, currentPage.bytes, currentPage.offset + currentPageOffset);
             this.currentPageOffset = currentPageOffset + 4;
         }
     }
@@ -189,7 +182,7 @@ public void writeIntLE(int i) throws IOException {
             super.writeIntLE(i);
         } else {
             BytesRef currentPage = currentBytesRef;
-            VH_LE_INT.set(currentPage.bytes, currentPage.offset + currentPageOffset, i);
+            ByteUtils.writeIntLE(i, currentPage.bytes, currentPage.offset + currentPageOffset);
             this.currentPageOffset = currentPageOffset + 4;
         }
     }
@@ -201,7 +194,7 @@ public void writeLong(long i) throws IOException {
             super.writeLong(i);
         } else {
             BytesRef currentPage = currentBytesRef;
-            VH_BE_LONG.set(currentPage.bytes, currentPage.offset + currentPageOffset, i);
+            ByteUtils.writeLongBE(i, currentPage.bytes, currentPage.offset + currentPageOffset);
             this.currentPageOffset = currentPageOffset + 8;
         }
     }
@@ -213,7 +206,7 @@ public void writeLongLE(long i) throws IOException {
             super.writeLongLE(i);
         } else {
             BytesRef currentPage = currentBytesRef;
-            VH_LE_LONG.set(currentPage.bytes, currentPage.offset + currentPageOffset, i);
+            ByteUtils.writeLongLE(i, currentPage.bytes, currentPage.offset + currentPageOffset);
             this.currentPageOffset = currentPageOffset + 8;
         }
     }
@@ -241,6 +234,25 @@ public void legacyWriteWithSizePrefix(Writeable writeable) throws IOException {
         }
     }
 
+    /**
+     * Attempt to get one page to perform a write directly into the page. The page will only be returned if the requested bytes can fit.
+     * If requested bytes cannot fit, null will be returned. This will advance the current position in the stream.
+     *
+     * @param bytes the number of bytes for the single write
+     * @return a direct page if there is enough space in current page, otherwise null
+     */
+    public BytesRef tryGetPageForWrite(int bytes) {
+        final int beforePageOffset = this.currentPageOffset;
+        if (bytes <= (pageSize - beforePageOffset)) {
+            BytesRef currentPage = currentBytesRef;
+            BytesRef bytesRef = new BytesRef(currentPage.bytes, currentPage.offset + beforePageOffset, bytes);
+            this.currentPageOffset = beforePageOffset + bytes;
+            return bytesRef;
+        } else {
+            return null;
+        }
+    }
+
     // overridden with some code duplication the same way other write methods in this class are overridden to bypass StreamOutput's
     // intermediary buffers
     @Override

server/src/main/java/org/elasticsearch/index/translog/OperationListener.java

@@ -9,18 +9,15 @@
 
 package org.elasticsearch.index.translog;
 
-import org.elasticsearch.common.bytes.BytesReference;
-
 @FunctionalInterface
 public interface OperationListener {
 
     /**
-     * This method is called when a new operation is added to the translog. The BytesReference is a releasable
-     * instance, so it should not be retained beyond the scope of this method.
+     * This method is called when a new operation is added to the translog.
      *
-     * @param data a releasable bytes reference of the data add
+     * @param operation the serialized operation added to the translog
      * @param seqNo the sequence number of the operation
      * @param location the location written
      */
-    void operationAdded(BytesReference data, long seqNo, Translog.Location location);
+    void operationAdded(Translog.Serialized operation, long seqNo, Translog.Location location);
 }