Add preload data support for putObject() API. (#1156)

When this flag is enabled, entire part/object data is loaded into memory to
enable connection retry on network failure in the middle of upload.

Author: balamurugana

api/src/main/java/io/minio/MinioClient.java

@@ -76,6 +76,7 @@
 import io.minio.messages.VersioningConfiguration;
 import io.minio.org.apache.commons.validator.routines.InetAddressValidator;
 import java.io.BufferedInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -626,9 +627,8 @@ protected Response execute(
     traceBuilder.append("\n");
 
     OkHttpClient httpClient = this.httpClient;
-    if (method == Method.PUT || method == Method.POST) {
-      // Issue #924: disable connection retry for PUT and POST methods. Its safe to do
-      // retry for other methods.
+    if (!(body instanceof byte[]) && (method == Method.PUT || method == Method.POST)) {
+      // Issue #924: disable connection retry for PUT and POST methods for other than byte array.
       httpClient = this.httpClient.newBuilder().retryOnConnectionFailure(false).build();
     }
 
@@ -2676,7 +2676,7 @@ public void removeBucket(RemoveBucketArgs args)
   }
 
   private ObjectWriteResponse putObject(
-      ObjectWriteArgs args,
+      PutObjectBaseArgs args,
       Object data,
       long objectSize,
       long partSize,
@@ -2694,6 +2694,7 @@ private ObjectWriteResponse putObject(
     String uploadId = null;
     long uploadedSize = 0L;
     Part[] parts = null;
+    ByteArrayOutputStream buffer = null;
 
     try {
       for (int partNumber = 1; partNumber <= partCount || partCount < 0; partNumber++) {
@@ -2714,12 +2715,19 @@ private ObjectWriteResponse putObject(
           }
         }
 
+        Object body = data;
+        if (args.preloadData()) {
+          if (buffer == null) buffer = new ByteArrayOutputStream();
+          fillData(buffer, data, availableSize);
+          body = buffer.toByteArray();
+        }
+
         if (partCount == 1) {
           return putObject(
               args.bucket(),
               args.region(),
               args.object(),
-              data,
+              body,
               (int) availableSize,
               headers,
               args.extraQueryParams());
@@ -2744,7 +2752,7 @@ private ObjectWriteResponse putObject(
                 args.bucket(),
                 args.region(),
                 args.object(),
-                data,
+                body,
                 (int) availableSize,
                 uploadId,
                 partNumber,
@@ -3807,6 +3815,25 @@ private long getAvailableSize(Object data, long expectedReadSize)
     return totalBytesRead;
   }
 
+  private void fillData(ByteArrayOutputStream buffer, Object data, long size) throws IOException {
+    buffer.reset();
+    byte[] buf = new byte[16384]; // 16KiB buffer for optimization
+    long totalBytesRead = 0;
+    while (totalBytesRead < size) {
+      long bytesToRead = size - totalBytesRead;
+      if (bytesToRead > buf.length) bytesToRead = buf.length;
+      int bytesRead = -1;
+      if (data instanceof RandomAccessFile) {
+        bytesRead = ((RandomAccessFile) data).read(buf, 0, (int) bytesToRead);
+      } else if (data instanceof BufferedInputStream) {
+        bytesRead = ((BufferedInputStream) data).read(buf, 0, (int) bytesToRead);
+      }
+      if (bytesRead < 0) throw new IOException("EOF on reading data");
+      buffer.write(buf, 0, bytesRead);
+      totalBytesRead += bytesRead;
+    }
+  }
+
   /**
    * Sets HTTP connect, write and read timeouts. A value of 0 means no timeout, otherwise values
    * must be between 1 and Integer.MAX_VALUE when converted to milliseconds.

api/src/main/java/io/minio/PutObjectArgs.java

@@ -43,7 +43,7 @@ public static Builder builder() {
   }
 
   /** Argument builder of {@link PutObjectArgs}. */
-  public static final class Builder extends ObjectWriteArgs.Builder<Builder, PutObjectArgs> {
+  public static final class Builder extends PutObjectBaseArgs.Builder<Builder, PutObjectArgs> {
     @Override
     protected void validate(PutObjectArgs args) {
       super.validate(args);

api/src/main/java/io/minio/PutObjectBaseArgs.java

@@ -25,6 +25,7 @@ public abstract class PutObjectBaseArgs extends ObjectWriteArgs {
   protected long partSize;
   protected int partCount;
   protected String contentType;
+  protected boolean preloadData;
 
   public long objectSize() {
     return objectSize;
@@ -51,9 +52,24 @@ public String contentType() throws IOException {
     return null;
   }
 
+  public boolean preloadData() {
+    return preloadData;
+  }
+
   /** Base argument builder class for {@link PutObjectBaseArgs}. */
+  @SuppressWarnings("unchecked") // Its safe to type cast to B as B is inherited by this class
   public abstract static class Builder<B extends Builder<B, A>, A extends PutObjectBaseArgs>
-      extends ObjectWriteArgs.Builder<B, A> {}
+      extends ObjectWriteArgs.Builder<B, A> {
+    /**
+     * Sets flag to control data preload of stream/file. When this flag is enabled, entire
+     * part/object data is loaded into memory to enable connection retry on network failure in the
+     * middle of upload.
+     */
+    public B preloadData(boolean preloadData) {
+      operations.add(args -> args.preloadData = preloadData);
+      return (B) this;
+    }
+  }
 
   @Override
   public boolean equals(Object o) {
@@ -64,11 +80,13 @@ public boolean equals(Object o) {
     return objectSize == that.objectSize
         && partSize == that.partSize
         && partCount == that.partCount
-        && Objects.equal(contentType, that.contentType);
+        && Objects.equal(contentType, that.contentType)
+        && preloadData == that.preloadData;
   }
 
   @Override
   public int hashCode() {
-    return Objects.hashCode(super.hashCode(), objectSize, partSize, partCount, contentType);
+    return Objects.hashCode(
+        super.hashCode(), objectSize, partSize, partCount, contentType, preloadData);
   }
 }

api/src/main/java/io/minio/UploadObjectArgs.java

@@ -50,7 +50,7 @@ public static Builder builder() {
   }
 
   /** Argument builder of {@link UploadObjectArgs}. */
-  public static final class Builder extends ObjectWriteArgs.Builder<Builder, UploadObjectArgs> {
+  public static final class Builder extends PutObjectBaseArgs.Builder<Builder, UploadObjectArgs> {
     @Override
     protected void validate(UploadObjectArgs args) {
       super.validate(args);