public Map<String, Long> getEtlFilePaths(String outputPath, BrokerDesc brokerDesc) throws Exception {
        Map<String, Long> filePathToSize = Maps.newHashMap();

        List<TBrokerFileStatus> fileStatuses = Lists.newArrayList();
        String etlFilePaths = outputPath + "/*";
        try {
            if (brokerDesc.hasBroker()) {
                BrokerUtil.parseFile(etlFilePaths, brokerDesc, fileStatuses);
            } else {
                HdfsUtil.parseFile(etlFilePaths, brokerDesc, fileStatuses);
            }
        } catch (UserException e) {
            throw new Exception(e);
        }

        for (TBrokerFileStatus fstatus : fileStatuses) {
            if (fstatus.isDir) {
                continue;
            }
            filePathToSize.put(fstatus.getPath(), fstatus.getSize());
        }
        LOG.debug("get spark etl file paths. files map: {}", filePathToSize);

        return filePathToSize;
    }
@Test
    public void testGetEtlFilePaths(@Mocked TFileBrokerService.Client client, @Mocked GlobalStateMgr globalStateMgr,
                                    @Injectable BrokerMgr brokerMgr) throws Exception {
        // list response
        TBrokerListResponse response = new TBrokerListResponse();
        TBrokerOperationStatus status = new TBrokerOperationStatus();
        status.statusCode = TBrokerOperationStatusCode.OK;
        response.opStatus = status;
        List<TBrokerFileStatus> files = Lists.newArrayList();
        String filePath = "hdfs://127.0.0.1:10000/starrocks/jobs/1/label6/9/label6.10.11.12.0.666666.parquet";
        files.add(new TBrokerFileStatus(filePath, false, 10, false));
        response.files = files;

        FsBroker fsBroker = new FsBroker("127.0.0.1", 99999);

        new MockUp<ThriftConnectionPool<TFileBrokerService.Client>>() {
            @Mock
            public TFileBrokerService.Client borrowObject(TNetworkAddress address, int timeoutMs) throws Exception {
                return client;
            }

            @Mock
            public void returnObject(TNetworkAddress address, TFileBrokerService.Client object) {
            }

            @Mock
            public void invalidateObject(TNetworkAddress address, TFileBrokerService.Client object) {
            }
        };

        new Expectations() {
            {
                globalStateMgr.getBrokerMgr();
                result = brokerMgr;
                brokerMgr.getBroker(anyString, anyString);
                result = fsBroker;
            }
        };

        try (MockedStatic<ThriftRPCRequestExecutor> thriftConnectionPoolMockedStatic =
                     Mockito.mockStatic(ThriftRPCRequestExecutor.class)) {
            thriftConnectionPoolMockedStatic.when(()
                            -> ThriftRPCRequestExecutor.call(Mockito.any(), Mockito.any(), Mockito.any()))
                    .thenReturn(response);

            BrokerDesc brokerDesc = new BrokerDesc(broker, Maps.newHashMap());
            SparkEtlJobHandler handler = new SparkEtlJobHandler();
            Map<String, Long> filePathToSize = handler.getEtlFilePaths(etlOutputPath, brokerDesc);
            Assert.assertTrue(filePathToSize.containsKey(filePath));
            Assert.assertEquals(10, (long) filePathToSize.get(filePath));
        }
    }