Change

Author: Tim-Brooks

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

@@ -11,9 +11,7 @@
 
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefIterator;
-import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.ByteUtils;
 
 import java.io.IOException;
@@ -224,11 +222,7 @@ public StreamInput streamInput() {
 
     @Override
     public void writeTo(OutputStream os) throws IOException {
-        if (os instanceof StreamOutput output) {
-            output.writeBytes(bytes, offset, length);
-        } else {
-            os.write(bytes, offset, length);
-        }
+        os.write(bytes, offset, length);
     }
 
     @Override

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

@@ -16,7 +16,6 @@
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.breaker.CircuitBreakingException;
 import org.elasticsearch.common.breaker.PreallocatedCircuitBreakerService;
-import org.elasticsearch.common.io.stream.RecyclerBytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.recycler.Recycler;
@@ -25,7 +24,6 @@
 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;
@@ -472,7 +470,6 @@ public T getAndSet(long index, T value) {
 
     @Nullable
     final PageCacheRecycler recycler;
-    final BytesRefRecycler bytesRefRecycler;
     @Nullable
     private final CircuitBreakerService breakerService;
     @Nullable
@@ -494,7 +491,6 @@ 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);
@@ -592,10 +588,6 @@ private <T extends BigArray> T validate(T array) {
         return array;
     }
 
-    public RecyclerBytesStreamOutput newRecyclerStreamOutput() {
-        return new RecyclerBytesStreamOutput(bytesRefRecycler);
-    }
-
     /**
      * Allocate a new {@link ByteArray}.
      * @param size          the initial length of the array

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

@@ -17,7 +17,7 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.DiskIoBufferPool;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
-import org.elasticsearch.common.io.stream.RecyclerBytesStreamOutput;
+import org.elasticsearch.common.io.stream.ReleasableBytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -608,7 +608,7 @@ TranslogWriter createWriter(
      * @throws IOException if adding the operation to the translog resulted in an I/O exception
      */
     public Location add(final Operation operation) throws IOException {
-        try (RecyclerBytesStreamOutput out = bigArrays.newRecyclerStreamOutput()) {
+        try (ReleasableBytesStreamOutput out = new ReleasableBytesStreamOutput(bigArrays)) {
             writeOperationWithSize(out, operation);
             final BytesReference bytes = out.bytes();
             readLock.lock();
@@ -1640,7 +1640,7 @@ public static void writeOperationNoSize(BufferedChecksumStreamOutput out, Transl
         out.writeInt((int) checksum);
     }
 
-    public static void writeOperationWithSize(RecyclerBytesStreamOutput out, Translog.Operation op) throws IOException {
+    public static void writeOperationWithSize(BytesStreamOutput out, Translog.Operation op) throws IOException {
         final long start = out.position();
         out.skip(Integer.BYTES);
         writeOperationNoSize(new BufferedChecksumStreamOutput(out), op);

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

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

server/src/main/java/org/elasticsearch/ingest/ESONFlat.java

@@ -96,43 +96,49 @@ public BytesReference getSerializedKeyBytes() {
         return serializedKeyBytes.get();
     }
 
-    private static final ThreadLocal<byte[]> BYTES_REF = ThreadLocal.withInitial(() -> new byte[16384]);
+    private static final ThreadLocal<BytesRef> BYTES_REF = ThreadLocal.withInitial(() -> new BytesRef(new byte[16384]));
 
     public static Recycler<BytesRef> getBytesRefRecycler() {
-        return new Recycler<>() {
+        return new ThreadLocalRecycler();
+    }
 
-            private boolean first = false;
+    private static class ThreadLocalRecycler implements Recycler<BytesRef> {
 
-            @Override
-            public V<BytesRef> obtain() {
-                final BytesRef bytesRef;
-                if (first) {
-                    first = false;
-                    bytesRef = new BytesRef(BYTES_REF.get());
-                } else {
-                    bytesRef = new BytesRef(new byte[16384]);
-                }
-                return new V<>() {
+        private boolean first = true;
 
-                    @Override
-                    public BytesRef v() {
-                        return bytesRef;
-                    }
+        @Override
+        public V<BytesRef> obtain() {
+            final BytesRef bytesRef;
+            if (first) {
+                first = false;
+                bytesRef = BYTES_REF.get();
+                bytesRef.offset = 0;
+                bytesRef.length = bytesRef.bytes.length;
+            } else {
+                bytesRef = new BytesRef(new byte[16384]);
+            }
+            return new VImpl(bytesRef);
+        }
 
-                    @Override
-                    public boolean isRecycled() {
-                        return false;
-                    }
+        private record VImpl(BytesRef bytesRef) implements V<BytesRef> {
 
-                    @Override
-                    public void close() {}
-                };
+            @Override
+            public BytesRef v() {
+                return bytesRef;
             }
 
             @Override
-            public int pageSize() {
-                return 16384;
+            public boolean isRecycled() {
+                return false;
             }
-        };
+
+            @Override
+            public void close() {}
+        }
+
+        @Override
+        public int pageSize() {
+            return 16384;
+        }
     }
 }