Change

Author: Tim-Brooks

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

@@ -25,6 +25,7 @@
 import java.nio.ByteOrder;
 import java.util.ArrayList;
 import java.util.Objects;
+import java.util.zip.Checksum;
 
 /**
  * A @link {@link StreamOutput} that uses {@link Recycler.V<BytesRef>} to acquire pages of bytes, which
@@ -348,6 +349,42 @@ public BytesReference bytes() {
         }
     }
 
+    public void calculateChecksum(Checksum checksum, int startPosition) {
+        int position = (int) position();
+        Objects.checkIndex(startPosition, position);
+
+        int bytesToProcess = position - startPosition;
+        if (bytesToProcess == 0) {
+            return;
+        }
+
+        int startPageIndex = startPosition / pageSize;
+        int startPageOffset = startPosition % pageSize;
+
+        final int remainder = position % pageSize;
+        final int bytesInLastPage = remainder != 0 ? remainder : pageSize;
+        final int endPageIndex = (position - 1) / pageSize;
+
+        if (startPageIndex == endPageIndex) {
+            BytesRef page = pages.get(startPageIndex).v();
+            checksum.update(page.bytes, page.offset + startPageOffset, bytesToProcess);
+        } else {
+            BytesRef firstPage = pages.get(startPageIndex).v();
+            int firstPageBytes = pageSize - startPageOffset;
+            checksum.update(firstPage.bytes, firstPage.offset + startPageOffset, firstPageBytes);
+
+            for (int i = startPageIndex + 1; i < endPageIndex; i++) {
+                BytesRef page = pages.get(i).v();
+                checksum.update(page.bytes, page.offset, pageSize);
+            }
+
+            if (endPageIndex > startPageIndex) {
+                BytesRef lastPage = pages.get(endPageIndex).v();
+                checksum.update(lastPage.bytes, lastPage.offset, bytesInLastPage);
+            }
+        }
+    }
+
     private void ensureCapacity(int bytesNeeded) {
         assert bytesNeeded > pageSize - currentPageOffset;
         ensureCapacityFromPosition(position() + bytesNeeded);

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;

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

@@ -16,7 +16,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.ReleasableBytesStreamOutput;
+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.io.stream.Writeable;
@@ -63,6 +63,7 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
+import java.util.zip.CRC32;
 
 import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.index.translog.TranslogConfig.EMPTY_TRANSLOG_BUFFER_SIZE;
@@ -610,7 +611,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 (ReleasableBytesStreamOutput out = new ReleasableBytesStreamOutput(bigArrays)) {
+        try (RecyclerBytesStreamOutput out = new RecyclerBytesStreamOutput(bigArrays.bytesRefRecycler())) {
             writeOperationWithSize(out, operation);
             final BytesReference bytes = out.bytes();
             readLock.lock();
@@ -1642,10 +1643,13 @@ public static void writeOperationNoSize(BufferedChecksumStreamOutput out, Transl
         out.writeInt((int) checksum);
     }
 
-    public static void writeOperationWithSize(BytesStreamOutput out, Translog.Operation op) throws IOException {
+    public static void writeOperationWithSize(RecyclerBytesStreamOutput out, Translog.Operation op) throws IOException {
         final long start = out.position();
         out.skip(Integer.BYTES);
-        writeOperationNoSize(new BufferedChecksumStreamOutput(out), op);
+        op.writeTo(out);
+        CRC32 checksum = new CRC32();
+        out.calculateChecksum(checksum, Math.toIntExact(start + 4));
+        out.writeInt((int) checksum.getValue());
         final long end = out.position();
         final int operationSize = (int) (end - Integer.BYTES - start);
         out.seek(start);