Use recycler bytestream output for translog writer (elastic#135161)

Use the recycler bytestream output for translog writer. This variant
allows direct access to the current page and variable page sizes which
will additional optimizations down the line.

Author: Tim-Brooks

server/src/main/java/org/elasticsearch/common/util/BigArrays.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.Streams;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
+import org.elasticsearch.transport.BytesRefRecycler;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -470,6 +471,7 @@ public T getAndSet(long index, T value) {
 
     @Nullable
     final PageCacheRecycler recycler;
+    final BytesRefRecycler bytesRefRecycler;
     @Nullable
     private final CircuitBreakerService breakerService;
     @Nullable
@@ -491,6 +493,7 @@ protected BigArrays(
     ) {
         this.checkBreaker = checkBreaker;
         this.recycler = recycler;
+        this.bytesRefRecycler = recycler != null ? new BytesRefRecycler(recycler) : BytesRefRecycler.NON_RECYCLING_INSTANCE;
         this.breakerService = breakerService;
         if (breakerService != null) {
             breaker = breakerService.getBreaker(breakerName);
@@ -588,6 +591,10 @@ private <T extends BigArray> T validate(T array) {
         return array;
     }
 
+    public BytesRefRecycler bytesRefRecycler() {
+        return bytesRefRecycler;
+    }
+
     /**
      * Allocate a new {@link ByteArray}.
      * @param size          the initial length of the array

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

@@ -17,7 +17,7 @@
 import org.elasticsearch.common.bytes.ReleasableBytesReference;
 import org.elasticsearch.common.io.Channels;
 import org.elasticsearch.common.io.DiskIoBufferPool;
-import org.elasticsearch.common.io.stream.ReleasableBytesStreamOutput;
+import org.elasticsearch.common.io.stream.RecyclerBytesStreamOutput;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.ReleasableLock;
@@ -82,7 +82,7 @@ public class TranslogWriter extends BaseTranslogReader implements Closeable {
     private List<Long> nonFsyncedSequenceNumbers = new ArrayList<>(64);
     private final int forceWriteThreshold;
     private volatile long bufferedBytes;
-    private ReleasableBytesStreamOutput buffer;
+    private RecyclerBytesStreamOutput buffer;
 
     private final Map<Long, Tuple<BytesReference, Exception>> seenSequenceNumbers;
 
@@ -245,7 +245,7 @@ public Translog.Location add(final BytesReference data, final long seqNo) throws
         synchronized (this) {
             ensureOpen();
             if (buffer == null) {
-                buffer = new ReleasableBytesStreamOutput(bigArrays);
+                buffer = new RecyclerBytesStreamOutput(bigArrays.bytesRefRecycler());
             }
             assert bufferedBytes == buffer.size();
             final long offset = totalOffset;
@@ -544,10 +544,11 @@ private void writeBufferedOps(long offset, boolean blockOnExistingWriter) throws
     private synchronized ReleasableBytesReference pollOpsToWrite() {
         ensureOpen();
         if (this.buffer != null) {
-            ReleasableBytesStreamOutput toWrite = this.buffer;
-            this.buffer = null;
-            this.bufferedBytes = 0;
-            return new ReleasableBytesReference(toWrite.bytes(), toWrite);
+            try (RecyclerBytesStreamOutput toWrite = this.buffer) {
+                this.buffer = null;
+                this.bufferedBytes = 0;
+                return toWrite.moveToBytesReference();
+            }
         } else {
             return ReleasableBytesReference.empty();
         }