Add FS based testing implementation for concurrent multipart uploads

Author: tlrx

modules/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureBlobContainer.java

@@ -15,7 +15,6 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.common.CheckedBiFunction;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobPath;
@@ -116,7 +115,7 @@ public void writeBlobAtomic(
         OperationPurpose purpose,
         String blobName,
         long blobSize,
-        CheckedBiFunction<Long, Long, InputStream, IOException> provider,
+        BlobMultiPartInputStreamProvider provider,
         boolean failIfAlreadyExists
     ) throws IOException {
         blobStore.writeBlobAtomic(purpose, buildKey(blobName), blobSize, provider, failIfAlreadyExists);

server/src/main/java/org/elasticsearch/common/blobstore/BlobContainer.java

@@ -153,11 +153,44 @@ default boolean supportsConcurrentMultipartUploads() {
         return false;
     }
 
+    /**
+     * Provides an {@link InputStream} to read a part of the blob content.
+     */
+    @FunctionalInterface
+    interface BlobMultiPartInputStreamProvider extends CheckedBiFunction<Long, Long, InputStream, IOException> {
+        /**
+         * Provides an {@link InputStream} to read a part of the blob content.
+         *
+         * @param offset        the offset in the blob content to start reading bytes from
+         * @param length        the number of bytes to read
+         * @return              an {@link InputStream} to read a part of the blob content.
+         * @throws IOException  if something goes wrong opening the input stream
+         */
+        @Override
+        InputStream apply(Long offset, Long length) throws IOException;
+    }
+
+    /**
+     * Reads the blob's content by calling an input stream provider multiple times, in order to split the blob's content into multiple
+     * parts that can be written to the container concurrently before being assembled into the final blob, using an atomic write operation
+     * if the implementation supports it. The number and the size of the parts depends of the implementation.
+     *
+     * Note: the method {link {@link #supportsConcurrentMultipartUploads()}} must be checked before calling this method.
+     *
+     * @param purpose             The purpose of the operation
+     * @param blobName            The name of the blob to write the contents of the input stream to.
+     * @param provider            The input stream provider that is used to read the blob content
+     * @param blobSize            The size of the blob to be written, in bytes. Must be the amount of bytes in the input stream. It is
+     *                            implementation dependent whether this value is used in writing the blob to the repository.
+     * @param failIfAlreadyExists whether to throw a FileAlreadyExistsException if the given blob already exists
+     * @throws FileAlreadyExistsException if failIfAlreadyExists is true and a blob by the same name already exists
+     * @throws IOException                if the input stream could not be read, or the target blob could not be written to.
+     */
     default void writeBlobAtomic(
         OperationPurpose purpose,
         String blobName,
         long blobSize,
-        CheckedBiFunction<Long, Long, InputStream, IOException> provider,
+        BlobMultiPartInputStreamProvider provider,
         boolean failIfAlreadyExists
     ) throws IOException {
         throw new UnsupportedOperationException();

server/src/main/java/org/elasticsearch/common/blobstore/fs/FsBlobContainer.java

@@ -89,6 +89,10 @@ public FsBlobContainer(FsBlobStore blobStore, BlobPath blobPath, Path path) {
         this.path = path;
     }
 
+    public Path getPath() {
+        return path;
+    }
+
     @Override
     public Map<String, BlobMetadata> listBlobs(OperationPurpose purpose) throws IOException {
         return listBlobsByPrefix(purpose, null);

server/src/main/java/org/elasticsearch/common/blobstore/support/FilterBlobContainer.java

@@ -88,6 +88,22 @@ public void writeMetadataBlob(
         delegate.writeMetadataBlob(purpose, blobName, failIfAlreadyExists, atomic, writer);
     }
 
+    @Override
+    public boolean supportsConcurrentMultipartUploads() {
+        return delegate.supportsConcurrentMultipartUploads();
+    }
+
+    @Override
+    public void writeBlobAtomic(
+        OperationPurpose purpose,
+        String blobName,
+        long blobSize,
+        BlobMultiPartInputStreamProvider provider,
+        boolean failIfAlreadyExists
+    ) throws IOException {
+        delegate.writeBlobAtomic(purpose, blobName, blobSize, provider, failIfAlreadyExists);
+    }
+
     @Override
     public void writeBlobAtomic(
         OperationPurpose purpose,

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

@@ -440,7 +440,7 @@ public static String getRepositoryDataBlobName(long repositoryGeneration) {
 
     private final NamedXContentRegistry namedXContentRegistry;
 
-    protected final BigArrays bigArrays;
+    protected BigArrays bigArrays;
 
     /**
      * Flag that is set to {@code true} if this instance is started with {@link #metadata} that has a higher value for