introduce azure client

pom.xml

@@ -114,7 +114,7 @@
         <jettison.version>1.5.4</jettison.version>
         <!-- ob-loader-dumper involved -->
         <hadoop-common.version>3.3.6</hadoop-common.version>
-        <netty.version>4.1.94.Final</netty.version>
+        <netty.version>4.1.108.Final</netty.version>
         <obkv-table-client.version>1.2.1</obkv-table-client.version>
         <mina.version>2.1.6</mina.version>
 
@@ -956,6 +956,17 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
+            <dependency>
+                <groupId>com.azure</groupId>
+                <artifactId>azure-storage-blob</artifactId>
+                <version>12.29.0</version>
+            </dependency>
+            <dependency>
+                <groupId>com.azure</groupId>
+                <artifactId>azure-storage-common</artifactId>
+                <version>12.29.0</version>
+            </dependency>
+
             <dependency>
                 <groupId>com.oceanbase</groupId>
                 <artifactId>apsara-audit-spring-boot-starter</artifactId>

server/odc-core/src/main/java/com/oceanbase/odc/core/shared/constant/ConnectType.java

@@ -39,6 +39,8 @@ public enum ConnectType {
     OBS(DialectType.FILE_SYSTEM),
     COS(DialectType.FILE_SYSTEM),
     S3A(DialectType.FILE_SYSTEM),
+    // microsoft azure blob service
+    BLOB(DialectType.FILE_SYSTEM),
     UNKNOWN(DialectType.UNKNOWN),
 
     ;

server/odc-service/pom.xml

@@ -67,6 +67,14 @@
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.azure</groupId>
+            <artifactId>azure-storage-blob</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.azure</groupId>
+            <artifactId>azure-storage-common</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.micrometer</groupId>
             <artifactId>micrometer-registry-prometheus</artifactId>

server/odc-service/src/main/java/com/oceanbase/odc/service/connection/FileSystemConnectionTester.java

@@ -32,6 +32,7 @@
 import com.oceanbase.odc.service.cloud.model.CloudProvider;
 import com.oceanbase.odc.service.connection.model.ConnectionConfig;
 import com.oceanbase.odc.service.connection.model.ConnectionTestResult;
+import com.oceanbase.odc.service.loaddata.model.ObjectStorageConfig;
 import com.oceanbase.odc.service.objectstorage.cloud.CloudResourceConfigurations;
 import com.oceanbase.odc.service.objectstorage.cloud.client.CloudClient;
 import com.oceanbase.odc.service.objectstorage.cloud.client.CloudException;
@@ -70,7 +71,7 @@ public ConnectionTestResult test(@NonNull ConnectionConfig config) {
         storageConfig.setBucketName(uri.getAuthority());
         storageConfig.setRegion(config.getRegion());
         storageConfig.setCloudProvider(getCloudProvider(config.getType()));
-        storageConfig.setPublicEndpoint(getEndPointByRegion(config.getType(), config.getRegion()));
+        storageConfig.setPublicEndpoint(getEndPoint(config.getType(), config.getRegion(), config.getUsername()));
         try {
             CloudClient cloudClient =
                     new CloudResourceConfigurations.CloudClientBuilder().generateCloudClient(storageConfig);
@@ -123,7 +124,7 @@ private CloudProvider getCloudProvider(ConnectType type) {
         }
     }
 
-    private static String getEndPointByRegion(ConnectType type, String region) {
+    private static String getEndPoint(ConnectType type, String region, String userName) {
         switch (type) {
             case COS:
                 return MessageFormat.format(COS_ENDPOINT_PATTERN, region);
@@ -137,6 +138,8 @@ private static String getEndPointByRegion(ConnectType type, String region) {
                     return MessageFormat.format(S3_ENDPOINT_CN_PATTERN, region);
                 }
                 return MessageFormat.format(S3_ENDPOINT_GLOBAL_PATTERN, region);
+            case BLOB:
+                return ObjectStorageConfig.concatEndpoint(CloudProvider.AZURE, userName);
             default:
                 throw new IllegalArgumentException("regionToEndpoint is not applicable for storageType " + type);
         }

server/odc-service/src/main/java/com/oceanbase/odc/service/loaddata/model/ObjectStorageConfig.java

@@ -189,6 +189,7 @@ public String getEndpoint() {
         // while azure use the account name as one component in endpoint.
         // GCS need neither region nor access key to concat the endpoint.
         if (cloudProvider == CloudProvider.AZURE) {
+            // for azure, accessKey is account name, secretKey is the secret key
             this.endpoint = concatEndpoint(cloudProvider, accessKey);
         } else if (cloudProvider == CloudProvider.GOOGLE_CLOUD) {
             this.endpoint = GCS_ENDPOINT;
@@ -217,7 +218,7 @@ public ObjectStorageConfiguration toObjectStorageConfiguration() {
         return config;
     }
 
-    private static String concatEndpoint(CloudProvider cloudProvider, String component) {
+    public static String concatEndpoint(CloudProvider cloudProvider, String component) {
         switch (cloudProvider) {
             case TENCENT_CLOUD:
                 return MessageFormat.format(COS_ENDPOINT_PATTERN, component);

server/odc-service/src/main/java/com/oceanbase/odc/service/loaddata/util/CloudProviderUtil.java

@@ -46,24 +46,4 @@ public static CloudProvider fromScheme(String scheme) {
                 throw new IllegalArgumentException("Unsupported scheme: " + scheme);
         }
     }
-
-    public static String getObjectUri(CloudProvider provider, String bucket, String objectName) {
-        switch (provider) {
-            case ALIBABA_CLOUD:
-                return "oss://" + bucket + "/" + objectName;
-            case TENCENT_CLOUD:
-                return "cos://" + bucket + "/" + objectName;
-            case HUAWEI_CLOUD:
-                return "obs://" + bucket + "/" + objectName;
-            case AWS:
-                return "s3://" + bucket + "/" + objectName;
-            case AZURE:
-                return "azblob://" + bucket + "/" + objectName;
-            case GOOGLE_CLOUD:
-                return "gcs://" + bucket + "/" + objectName;
-            default:
-                throw new IllegalArgumentException("Unsupported cloud provider: " + provider);
-        }
-    }
-
 }

server/odc-service/src/main/java/com/oceanbase/odc/service/objectstorage/client/CloudObjectStorageClient.java

@@ -290,7 +290,7 @@ private CompleteMultipartUploadResult multiPartUpload(@NotBlank String objectNam
             }
         }
         CompleteMultipartUploadRequest completeMultipartUploadRequest =
-                new CompleteMultipartUploadRequest(bucketName, objectName, uploadId, partTags);
+                new CompleteMultipartUploadRequest(bucketName, objectName, uploadId, partTags, metadata);
         CompleteMultipartUploadResult completeMultipartUploadResult =
                 internalEndpointCloudObjectStorage.completeMultipartUpload(completeMultipartUploadRequest);
         log.info("Complete multipart upload, result={}", completeMultipartUploadResult);

server/odc-service/src/main/java/com/oceanbase/odc/service/objectstorage/cloud/CloudResourceConfigurations.java

@@ -39,11 +39,15 @@
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
 import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
 import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClientBuilder;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+import com.azure.storage.common.StorageSharedKeyCredential;
 import com.oceanbase.odc.common.util.StringUtils;
 import com.oceanbase.odc.core.shared.PreConditions;
 import com.oceanbase.odc.service.cloud.model.CloudProvider;
 import com.oceanbase.odc.service.objectstorage.cloud.client.AlibabaCloudClient;
 import com.oceanbase.odc.service.objectstorage.cloud.client.AmazonCloudClient;
+import com.oceanbase.odc.service.objectstorage.cloud.client.AzureCloudClient;
 import com.oceanbase.odc.service.objectstorage.cloud.client.CloudClient;
 import com.oceanbase.odc.service.objectstorage.cloud.client.GoogleCloudClient;
 import com.oceanbase.odc.service.objectstorage.cloud.client.NullCloudClient;
@@ -110,6 +114,8 @@ public CloudClient generateCloudClient(ObjectStorageConfiguration configuration,
                     return createAmazonCloudClient(configuration);
                 case GOOGLE_CLOUD:
                     return createGoogleCloudClient(configuration);
+                case AZURE:
+                    return createAzureCloudClient(configuration);
                 default:
                     return new NullCloudClient();
             }
@@ -198,6 +204,18 @@ static GoogleCloudClient createGoogleCloudClient(ObjectStorageConfiguration conf
         return new GoogleCloudClient(s3, sts, roleSessionName, roleArn);
     }
 
+    static AzureCloudClient createAzureCloudClient(ObjectStorageConfiguration configuration) {
+        // for azure, accessKey is account name, secretKey is the secret key
+        String accountName = configuration.getAccessKeyId();
+        String accountKey = configuration.getAccessKeySecret();
+        StorageSharedKeyCredential credential = new StorageSharedKeyCredential(accountName, accountKey);
+        BlobServiceClient serviceClient = new BlobServiceClientBuilder()
+                .endpoint(configuration.getPublicEndpoint())
+                .credential(credential)
+                .buildClient();
+        return new AzureCloudClient(serviceClient, configuration.getRegion());
+    }
+
     private static OSS getInternalOss(ObjectStorageConfiguration objectStorageConfiguration) {
         return createOssClient(
                 objectStorageConfiguration.getAccessKeyId(),