public List<TBrokerFileStatus> listPath(String path, boolean fileNameOnly, Map<String, String> properties) {
        List<TBrokerFileStatus> resultFileStatus = null;
        WildcardURI pathUri = new WildcardURI(path);
        BrokerFileSystem fileSystem = getFileSystem(path, properties);
        Path pathPattern = new Path(pathUri.getPath());
        try {
            FileStatus[] files = fileSystem.getDFSFileSystem().globStatus(pathPattern);
            if (files == null) {
                resultFileStatus = new ArrayList<>(0);
                return resultFileStatus;
            }
            resultFileStatus = new ArrayList<>(files.length);
            for (FileStatus fileStatus : files) {
                TBrokerFileStatus brokerFileStatus = new TBrokerFileStatus();
                brokerFileStatus.setIsDir(fileStatus.isDirectory());
                if (fileStatus.isDirectory()) {
                    brokerFileStatus.setIsSplitable(false);
                    brokerFileStatus.setSize(-1);
                } else {
                    brokerFileStatus.setSize(fileStatus.getLen());
                    brokerFileStatus.setIsSplitable(true);
                }
                if (fileNameOnly) {
                    // return like this: file.txt
                    brokerFileStatus.setPath(fileStatus.getPath().getName());
                } else {
                    // return like this: //path/to/your/file.txt
                    brokerFileStatus.setPath(fileStatus.getPath().toString());
                }
                resultFileStatus.add(brokerFileStatus);
            }
        } catch (FileNotFoundException e) {
            logger.info("file not found: " + e.getMessage());
            throw new BrokerException(TBrokerOperationStatusCode.FILE_NOT_FOUND,
                    e, "file not found");
        } catch (IllegalArgumentException e) {
            logger.error("The arguments of blob store(S3/Azure) may be wrong. You can check " +
                      "the arguments like region, IAM, instance profile and so on.");
            throw new BrokerException(TBrokerOperationStatusCode.TARGET_STORAGE_SERVICE_ERROR,
                e, "The arguments of blob store(S3/Azure) may be wrong. " +
                   "You can check the arguments like region, IAM, " +
                   "instance profile and so on.");
        } catch (Exception e) {
            logger.error("errors while get file status ", e);
            throw new BrokerException(TBrokerOperationStatusCode.TARGET_STORAGE_SERVICE_ERROR,
                    e, "unknown error when get file status");
        }
        return resultFileStatus;
    }
@Test
    public void testListPaths() {
        Map<String, String> properties = new HashMap<String, String>();
        properties.put("username", "user");
        properties.put("password", "passwd");
        
        List<TBrokerFileStatus> files2 = fileSystemManager.listPath(testHdfsHost + "/data/abc/logs/*.out",
                false, properties);
        assertEquals(files2.size(), 2);
    }