public static Map<String, String> getParamsFromFileStoreInfo(FileStoreInfo fsInfo) {
        Map<String, String> params = new HashMap<>();
        switch (fsInfo.getFsType()) {
            case S3:
                S3FileStoreInfo s3FileStoreInfo = fsInfo.getS3FsInfo();
                params.put(CloudConfigurationConstants.AWS_S3_REGION, s3FileStoreInfo.getRegion());
                params.put(CloudConfigurationConstants.AWS_S3_ENDPOINT, s3FileStoreInfo.getEndpoint());
                if (s3FileStoreInfo.getPartitionedPrefixEnabled()) {
                    // Don't show the parameters if not enabled.
                    params.put(CloudConfigurationConstants.AWS_S3_ENABLE_PARTITIONED_PREFIX,
                            Boolean.toString(true));
                    params.put(CloudConfigurationConstants.AWS_S3_NUM_PARTITIONED_PREFIX,
                            Integer.toString(s3FileStoreInfo.getNumPartitionedPrefix()));
                }
                AwsCredentialInfo credentialInfo = s3FileStoreInfo.getCredential();
                if (credentialInfo.hasSimpleCredential()) {
                    params.put(CloudConfigurationConstants.AWS_S3_USE_INSTANCE_PROFILE, "false");
                    params.put(CloudConfigurationConstants.AWS_S3_USE_AWS_SDK_DEFAULT_BEHAVIOR, "false");
                    params.put(CloudConfigurationConstants.AWS_S3_ACCESS_KEY,
                            credentialInfo.getSimpleCredential().getAccessKey());
                    params.put(CloudConfigurationConstants.AWS_S3_SECRET_KEY,
                            credentialInfo.getSimpleCredential().getAccessKeySecret());
                } else if (credentialInfo.hasAssumeRoleCredential()) {
                    params.put(CloudConfigurationConstants.AWS_S3_USE_INSTANCE_PROFILE, "true");
                    params.put(CloudConfigurationConstants.AWS_S3_USE_AWS_SDK_DEFAULT_BEHAVIOR, "false");
                    params.put(CloudConfigurationConstants.AWS_S3_IAM_ROLE_ARN,
                            credentialInfo.getAssumeRoleCredential().getIamRoleArn());
                    params.put(CloudConfigurationConstants.AWS_S3_EXTERNAL_ID,
                            credentialInfo.getAssumeRoleCredential().getExternalId());
                } else if (credentialInfo.hasProfileCredential()) {
                    params.put(CloudConfigurationConstants.AWS_S3_USE_INSTANCE_PROFILE, "true");
                    params.put(CloudConfigurationConstants.AWS_S3_USE_AWS_SDK_DEFAULT_BEHAVIOR, "false");
                } else if (credentialInfo.hasDefaultCredential()) {
                    params.put(CloudConfigurationConstants.AWS_S3_USE_AWS_SDK_DEFAULT_BEHAVIOR, "true");
                }
                return params;
            case HDFS:
                HDFSFileStoreInfo hdfsFileStoreInfo = fsInfo.getHdfsFsInfo();
                params.putAll(hdfsFileStoreInfo.getConfigurationMap());
                String userName = hdfsFileStoreInfo.getUsername();
                if (!Strings.isNullOrEmpty(userName)) {
                    params.put(CloudConfigurationConstants.HDFS_USERNAME_DEPRECATED, userName);
                }
                return params;
            case AZBLOB:
                AzBlobFileStoreInfo azBlobFileStoreInfo = fsInfo.getAzblobFsInfo();
                params.put(CloudConfigurationConstants.AZURE_BLOB_ENDPOINT, azBlobFileStoreInfo.getEndpoint());
                AzBlobCredentialInfo azBlobcredentialInfo = azBlobFileStoreInfo.getCredential();
                String sharedKey = azBlobcredentialInfo.getSharedKey();
                if (!Strings.isNullOrEmpty(sharedKey)) {
                    params.put(CloudConfigurationConstants.AZURE_BLOB_SHARED_KEY, sharedKey);
                }
                String sasToken = azBlobcredentialInfo.getSasToken();
                if (!Strings.isNullOrEmpty(sasToken)) {
                    params.put(CloudConfigurationConstants.AZURE_BLOB_SAS_TOKEN, sasToken);
                }
                return params;
            default:
                return params;
        }
    }
@Test
    public void testGetParamsFromFileStoreInfo() {
        AwsCredentialInfo.Builder awsCredBuilder = AwsCredentialInfo.newBuilder();
        awsCredBuilder.getSimpleCredentialBuilder()
                .setAccessKey("ak")
                .setAccessKeySecret("sk")
                .build();

        FileStoreInfo.Builder fsInfoBuilder = FileStoreInfo.newBuilder();
        fsInfoBuilder.getS3FsInfoBuilder()
                .setBucket("bucket")
                .setEndpoint("endpoint")
                .setRegion("region")
                .setCredential(awsCredBuilder);

        fsInfoBuilder.setFsKey("0")
                .setFsType(FileStoreType.S3)
                .addLocations("s3://bucket");

        {
            FileStoreInfo fs = fsInfoBuilder.build();
            Map<String, String> params = StorageVolume.getParamsFromFileStoreInfo(fs);
            Assert.assertFalse(params.containsKey(CloudConfigurationConstants.AWS_S3_ENABLE_PARTITIONED_PREFIX));
            Assert.assertFalse(params.containsKey(CloudConfigurationConstants.AWS_S3_NUM_PARTITIONED_PREFIX));
        }

        fsInfoBuilder.getS3FsInfoBuilder()
                .setPartitionedPrefixEnabled(true)
                .setNumPartitionedPrefix(32);

        {
            FileStoreInfo fs = fsInfoBuilder.build();
            Map<String, String> params = StorageVolume.getParamsFromFileStoreInfo(fs);
            Assert.assertTrue(params.containsKey(CloudConfigurationConstants.AWS_S3_ENABLE_PARTITIONED_PREFIX));
            Assert.assertTrue(params.containsKey(CloudConfigurationConstants.AWS_S3_NUM_PARTITIONED_PREFIX));
            Assert.assertEquals("32", params.get(CloudConfigurationConstants.AWS_S3_NUM_PARTITIONED_PREFIX));
        }

        // It's OK to have trailing '/' after bucket name
        fsInfoBuilder.addLocations("s3://bucket/");
        {
            FileStoreInfo fs = fsInfoBuilder.build();
            ExceptionChecker.expectThrowsNoException(() -> StorageVolume.fromFileStoreInfo(fs));
        }

        // can't have more after bucket name
        fsInfoBuilder.addLocations("s3://bucket/abc");
        {
            FileStoreInfo fs = fsInfoBuilder.build();
            Assert.assertThrows(DdlException.class, () -> StorageVolume.fromFileStoreInfo(fs));
        }
    }