public BrokerFileSystem getFileSystem(String path, Map<String, String> properties) {
        WildcardURI pathUri = new WildcardURI(path);
        String scheme = pathUri.getUri().getScheme();
        if (Strings.isNullOrEmpty(scheme)) {
            throw new BrokerException(TBrokerOperationStatusCode.INVALID_INPUT_FILE_PATH,
                    "invalid path. scheme is null");
        }
        BrokerFileSystem brokerFileSystem = null;
        if (scheme.equals(HDFS_SCHEME) || scheme.equals(VIEWFS_SCHEME)) {
            brokerFileSystem = getDistributedFileSystem(scheme, path, properties);
        } else if (scheme.equals(S3A_SCHEME)) {
            brokerFileSystem = getS3AFileSystem(path, properties);
        } else if (scheme.equals(OSS_SCHEME)) {
            brokerFileSystem = getOSSFileSystem(path, properties);
        } else if (scheme.equals(COS_SCHEME)) {
            brokerFileSystem = getCOSFileSystem(path, properties);
        } else if (scheme.equals(KS3_SCHEME)) {
            brokerFileSystem = getKS3FileSystem(path, properties);
        } else if (scheme.equals(OBS_SCHEME)) {
            brokerFileSystem = getOBSFileSystem(path, properties);
        } else if (scheme.equals(TOS_SCHEME)) {
            brokerFileSystem = getTOSFileSystem(path, properties);
        } else {
            // If all above match fails, then we will read the settings from hdfs-site.xml, core-site.xml of FE,
            // and try to create a universal file system. The reason why we can do this is because hadoop/s3 
            // SDK is compatible with nearly all file/object storage system
            brokerFileSystem = getUniversalFileSystem(path, properties);
        }
        return brokerFileSystem;
    }
@Test
    public void testGetFileSystemForHAWithNoRpcConfig() throws IOException {
        Map<String, String> properties = new HashMap<String, String>();
        properties.put("username", "user");
        properties.put("password", "passwd");
        properties.put("fs.defaultFS", "hdfs://starrocks");
        properties.put("dfs.nameservices", "starrocks");
        properties.put("dfs.ha.namenodes.starrocks", "nn1,nn2");
        properties.put("dfs.namenode.rpc-address.starrocks.nn1", "host1:port1");
        properties.put("dfs.client.failover.proxy.provider.bdos",
                "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
        boolean haveException = false;
        try {
            BrokerFileSystem fs = fileSystemManager.getFileSystem(testHdfsHost + "/data/abc/logs", properties);
        } catch (BrokerException be) {
            haveException = true;
        }
        assertEquals(true, haveException);
    }