Improve volume upload concurrency and progress

sdk-bulkwriter/src/main/java/io/milvus/bulkwriter/VolumeBulkWriter.java

@@ -21,7 +21,6 @@
 
 import com.google.common.collect.Lists;
 import com.google.gson.JsonObject;
-import io.milvus.bulkwriter.common.clientenum.ConnectType;
 import io.milvus.bulkwriter.model.UploadFilesResult;
 import io.milvus.bulkwriter.request.volume.UploadFilesRequest;
 import io.milvus.common.utils.ExceptionUtils;
@@ -64,7 +63,7 @@ public VolumeBulkWriter(VolumeBulkWriterParam bulkWriterParam) throws IOExceptio
     private VolumeFileManager initVolumeFileManagerParams(VolumeBulkWriterParam bulkWriterParam) throws IOException {
         VolumeFileManagerParam volumeFileManagerParam = VolumeFileManagerParam.newBuilder()
                 .withCloudEndpoint(bulkWriterParam.getCloudEndpoint()).withApiKey(bulkWriterParam.getApiKey())
-                .withVolumeName(bulkWriterParam.getVolumeName()).withConnectType(ConnectType.AUTO)
+                .withVolumeName(bulkWriterParam.getVolumeName()).withConnectType(bulkWriterParam.getConnectType())
                 .build();
         return new VolumeFileManager(volumeFileManagerParam);
     }
@@ -149,9 +148,13 @@ protected void callBack(List<String> fileList) {
 
     @Override
     public void close() throws Exception {
-        logger.info("execute remaining actions to prevent loss of memory data or residual empty directories.");
-        exit();
-        logger.info(String.format("RemoteBulkWriter done! output remote files: %s", getBatchFiles()));
+        try {
+            logger.info("execute remaining actions to prevent loss of memory data or residual empty directories.");
+            exit();
+            logger.info(String.format("RemoteBulkWriter done! output remote files: %s", getBatchFiles()));
+        } finally {
+            volumeFileManager.shutdownGracefully();
+        }
     }
 
     private void serialImportData(List<String> fileList) {
@@ -199,6 +202,6 @@ private static String getMinioFilePath(String remotePath, String relativeFilePat
         relativeFilePath = relativeFilePath.startsWith("/") ? relativeFilePath.substring(1) : relativeFilePath;
         Path relative = Paths.get(relativeFilePath);
         Path joinedPath = remote.resolve(relative);
-        return joinedPath.toString();
+        return joinedPath.toString().replace("\\", "/");
     }
 }

sdk-bulkwriter/src/main/java/io/milvus/bulkwriter/VolumeBulkWriterParam.java

@@ -20,6 +20,7 @@
 package io.milvus.bulkwriter;
 
 import io.milvus.bulkwriter.common.clientenum.BulkFileType;
+import io.milvus.bulkwriter.common.clientenum.ConnectType;
 import io.milvus.bulkwriter.common.utils.V2AdapterUtils;
 import io.milvus.exception.ParamException;
 import io.milvus.param.ParamUtils;
@@ -42,6 +43,7 @@ public class VolumeBulkWriterParam {
     private final String cloudEndpoint;
     private final String apiKey;
     private final String volumeName;
+    private final ConnectType connectType;
 
     private VolumeBulkWriterParam(Builder builder) {
         this.collectionSchema = builder.collectionSchema;
@@ -53,6 +55,7 @@ private VolumeBulkWriterParam(Builder builder) {
         this.cloudEndpoint = builder.cloudEndpoint;
         this.apiKey = builder.apiKey;
         this.volumeName = builder.volumeName;
+        this.connectType = builder.connectType;
     }
 
     public CreateCollectionReq.CollectionSchema getCollectionSchema() {
@@ -87,6 +90,10 @@ public String getVolumeName() {
         return volumeName;
     }
 
+    public ConnectType getConnectType() {
+        return connectType;
+    }
+
     @Override
     public String toString() {
         return "VolumeBulkWriterParam{" +
@@ -96,6 +103,7 @@ public String toString() {
                 ", fileType=" + fileType +
                 ", cloudEndpoint='" + cloudEndpoint + '\'' +
                 ", volumeName='" + volumeName + '\'' +
+                ", connectType=" + connectType +
                 '}';
     }
 
@@ -118,6 +126,8 @@ public static final class Builder {
 
         private String volumeName;
 
+        private ConnectType connectType = ConnectType.AUTO;
+
         private Builder() {
         }
 
@@ -184,6 +194,11 @@ public Builder withVolumeName(String volumeName) {
             return this;
         }
 
+        public Builder withConnectType(ConnectType connectType) {
+            this.connectType = connectType;
+            return this;
+        }
+
         /**
          * Verifies parameters and creates a new {@link VolumeBulkWriterParam} instance.
          *
@@ -195,6 +210,9 @@ public VolumeBulkWriterParam build() throws ParamException {
             if (collectionSchema == null) {
                 throw new ParamException("collectionSchema cannot be null");
             }
+            ParamUtils.CheckNullEmptyString(cloudEndpoint, "cloudEndpoint");
+            ParamUtils.CheckNullEmptyString(apiKey, "apiKey");
+            ParamUtils.CheckNullEmptyString(volumeName, "volumeName");
 
             return new VolumeBulkWriterParam(this);
         }