Implement ccr file restore (#37130) (#37669)

This is related to #35975. It implements a file based restore in the
CcrRepository. The restore transfers files from the leader cluster
to the follower cluster. It does not implement any advanced resiliency
features at the moment. Any request failure will end the restore.

Author: Tim-Brooks

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpChannelTests.java

@@ -590,7 +590,7 @@ private static class TestResponse extends RestResponse {
             }
             final ByteArray bigArray = bigArrays.newByteArray(bytes.length);
             bigArray.set(0, bytes, 0, bytes.length);
-            reference = new ReleasablePagedBytesReference(bigArrays, bigArray, bytes.length, Releasables.releaseOnce(bigArray));
+            reference = new ReleasablePagedBytesReference(bigArray, bytes.length, Releasables.releaseOnce(bigArray));
         }
 
         @Override

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

@@ -21,7 +21,6 @@
 
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefIterator;
-import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.ByteArray;
 import org.elasticsearch.common.util.PageCacheRecycler;
 
@@ -35,17 +34,15 @@ public class PagedBytesReference extends BytesReference {
 
     private static final int PAGE_SIZE = PageCacheRecycler.BYTE_PAGE_SIZE;
 
-    private final BigArrays bigarrays;
-    protected final ByteArray byteArray;
+    private final ByteArray byteArray;
     private final int offset;
     private final int length;
 
-    public PagedBytesReference(BigArrays bigarrays, ByteArray byteArray, int length) {
-        this(bigarrays, byteArray, 0, length);
+    public PagedBytesReference(ByteArray byteArray, int length) {
+        this(byteArray, 0, length);
     }
 
-    public PagedBytesReference(BigArrays bigarrays, ByteArray byteArray, int from, int length) {
-        this.bigarrays = bigarrays;
+    private PagedBytesReference(ByteArray byteArray, int from, int length) {
         this.byteArray = byteArray;
         this.offset = from;
         this.length = length;
@@ -67,7 +64,7 @@ public BytesReference slice(int from, int length) {
             throw new IllegalArgumentException("can't slice a buffer with length [" + length() +
                 "], with slice parameters from [" + from + "], length [" + length + "]");
         }
-        return new PagedBytesReference(bigarrays, byteArray, offset + from, length);
+        return new PagedBytesReference(byteArray, offset + from, length);
     }
 
     @Override

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

@@ -21,7 +21,6 @@
 
 import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.common.lease.Releasables;
-import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.ByteArray;
 
 /**
@@ -32,9 +31,8 @@ public final class ReleasablePagedBytesReference extends PagedBytesReference imp
 
     private final Releasable releasable;
 
-    public ReleasablePagedBytesReference(BigArrays bigarrays, ByteArray byteArray, int length,
-                                         Releasable releasable) {
-        super(bigarrays, byteArray, length);
+    public ReleasablePagedBytesReference(ByteArray byteArray, int length, Releasable releasable) {
+        super(byteArray, length);
         this.releasable = releasable;
     }
 

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

@@ -140,7 +140,7 @@ public int size() {
 
     @Override
     public BytesReference bytes() {
-        return new PagedBytesReference(bigArrays, bytes, count);
+        return new PagedBytesReference(bytes, count);
     }
 
     /**

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

@@ -56,7 +56,7 @@ public ReleasableBytesStreamOutput(int expectedSize, BigArrays bigArrays) {
      */
     @Override
     public ReleasablePagedBytesReference bytes() {
-        return new ReleasablePagedBytesReference(bigArrays, bytes, count, releasable);
+        return new ReleasablePagedBytesReference(bytes, count, releasable);
     }
 
     @Override

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -120,8 +120,8 @@ public void testEquals() {
         }
 
         // get refs & compare
-        BytesReference pbr = new PagedBytesReference(bigarrays, ba1, length);
-        BytesReference pbr2 = new PagedBytesReference(bigarrays, ba2, length);
+        BytesReference pbr = new PagedBytesReference(ba1, length);
+        BytesReference pbr2 = new PagedBytesReference(ba2, length);
         assertEquals(pbr, pbr2);
         int offsetToFlip = randomIntBetween(0, length - 1);
         int value = ~Byte.toUnsignedInt(ba1.get(offsetToFlip));

server/src/main/java/org/elasticsearch/repositories/blobstore/FileRestoreContext.java

@@ -528,7 +528,7 @@ public void testEquals() throws IOException {
     public void testSliceEquals() {
         int length = randomIntBetween(100, PAGE_SIZE * randomIntBetween(2, 5));
         ByteArray ba1 = bigarrays.newByteArray(length, false);
-        BytesReference pbr = new PagedBytesReference(bigarrays, ba1, length);
+        BytesReference pbr = new PagedBytesReference(ba1, length);
 
         // test equality of slices
         int sliceFrom = randomIntBetween(0, pbr.length());

server/src/test/java/org/elasticsearch/common/bytes/PagedBytesReferenceTests.java

@@ -455,6 +455,12 @@ protected IndexShard newStartedShard(final boolean primary) throws IOException {
         return newStartedShard(primary, Settings.EMPTY);
     }
 
+    /**
+     * Creates a new empty shard and starts it.
+     *
+     * @param primary controls whether the shard will be a primary or a replica.
+     * @param settings the settings to use for this shard
+     */
     protected IndexShard newStartedShard(final boolean primary, Settings settings) throws IOException {
         return newStartedShard(primary, settings, new InternalEngineFactory());
     }