public static byte[] readFile(String path, BrokerDesc brokerDesc) throws UserException {
        TNetworkAddress address = getAddress(brokerDesc);
        TBrokerFD fd = null;
        try {
            // get file size
            TBrokerListPathRequest request = new TBrokerListPathRequest(
                    TBrokerVersion.VERSION_ONE, path, false, brokerDesc.getProperties());
            TBrokerListResponse tBrokerListResponse = ThriftRPCRequestExecutor.call(
                    ThriftConnectionPool.brokerPool,
                    address,
                    client -> client.listPath(request));

            if (tBrokerListResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
                throw new UserException("Broker list path failed. path=" + path + ", broker=" + address
                        + ",msg=" + tBrokerListResponse.getOpStatus().getMessage());
            }
            List<TBrokerFileStatus> fileStatuses = tBrokerListResponse.getFiles();
            if (fileStatuses.size() != 1) {
                throw new UserException("Broker files num error. path=" + path + ", broker=" + address
                        + ", files num: " + fileStatuses.size());
            }

            Preconditions.checkState(!fileStatuses.get(0).isIsDir());
            long fileSize = fileStatuses.get(0).getSize();

            // open reader
            String clientId = NetUtils.getHostPortInAccessibleFormat(
                    FrontendOptions.getLocalHostAddress(), Config.rpc_port);
            TBrokerOpenReaderRequest tOpenReaderRequest = new TBrokerOpenReaderRequest(
                    TBrokerVersion.VERSION_ONE, path, 0, clientId, brokerDesc.getProperties());
            TBrokerOpenReaderResponse tOpenReaderResponse = ThriftRPCRequestExecutor.call(
                    ThriftConnectionPool.brokerPool,
                    address,
                    client -> client.openReader(tOpenReaderRequest));

            if (tOpenReaderResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
                throw new UserException("Broker open reader failed. path=" + path + ", broker=" + address
                        + ", msg=" + tOpenReaderResponse.getOpStatus().getMessage());
            }
            fd = tOpenReaderResponse.getFd();

            // read
            TBrokerPReadRequest tPReadRequest = new TBrokerPReadRequest(
                    TBrokerVersion.VERSION_ONE, fd, 0, fileSize);
            TBrokerReadResponse tReadResponse = ThriftRPCRequestExecutor.call(
                    ThriftConnectionPool.brokerPool,
                    address,
                    client -> client.pread(tPReadRequest));

            if (tReadResponse.getOpStatus().getStatusCode() != TBrokerOperationStatusCode.OK) {
                throw new UserException("Broker read failed. path=" + path + ", broker=" + address
                        + ", msg=" + tReadResponse.getOpStatus().getMessage());
            }
            return tReadResponse.getData();
        } catch (TException e) {
            String failMsg = "Broker read file exception. path=" + path + ", broker=" + address;
            LOG.warn(failMsg, e);
            throw new UserException(failMsg);
        } finally {
            // close reader
            if (fd != null) {
                TBrokerOperationStatus tOperationStatus = null;
                try {
                    TBrokerCloseReaderRequest tCloseReaderRequest = new TBrokerCloseReaderRequest(
                            TBrokerVersion.VERSION_ONE, fd);
                    tOperationStatus = ThriftRPCRequestExecutor.call(
                            ThriftConnectionPool.brokerPool,
                            address,
                            client -> client.closeReader(tCloseReaderRequest));
                } catch (TException e) {
                    LOG.warn("Broker close reader failed. path={}, address={}", path, address, e);
                }
                if (tOperationStatus == null) {
                    LOG.warn("Broker close reader failed: operation status is null. path={}, address={}", path, address);
                } else if (tOperationStatus.getStatusCode() != TBrokerOperationStatusCode.OK) {
                    LOG.warn("Broker close reader failed. path={}, address={}, error={}", path, address,
                            tOperationStatus.getMessage());
                }
            }
        }
    }
@Test
    public void testReadFile(@Mocked TFileBrokerService.Client client, @Mocked GlobalStateMgr globalStateMgr,
                             @Injectable BrokerMgr brokerMgr)
            throws TException, UserException {
        // list response
        TBrokerListResponse listResponse = new TBrokerListResponse();
        TBrokerOperationStatus status = new TBrokerOperationStatus();
        status.statusCode = TBrokerOperationStatusCode.OK;
        listResponse.opStatus = status;
        List<TBrokerFileStatus> files = Lists.newArrayList();
        String filePath = "hdfs://127.0.0.1:10000/starrocks/jobs/1/label6/9/dpp_result.json";
        files.add(new TBrokerFileStatus(filePath, false, 10, false));
        listResponse.files = files;

        // open reader response
        TBrokerOpenReaderResponse openReaderResponse = new TBrokerOpenReaderResponse();
        openReaderResponse.opStatus = status;
        openReaderResponse.fd = new TBrokerFD(1, 2);

        // read response
        String dppResultStr = "{'normal_rows': 10, 'abnormal_rows': 0, 'failed_reason': 'etl job failed'}";
        TBrokerReadResponse readResponse = new TBrokerReadResponse();
        readResponse.opStatus = status;
        readResponse.setData(dppResultStr.getBytes(StandardCharsets.UTF_8));

        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) {
                return;
            }

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

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

            BrokerDesc brokerDesc = new BrokerDesc("broker0", Maps.newHashMap());
            try {
                byte[] data = BrokerUtil.readFile(filePath, brokerDesc);
                String readStr = new String(data, StandardCharsets.UTF_8);
                Assert.assertEquals(dppResultStr, readStr);
            } catch (Exception e) {
                Assert.fail(e.getMessage());
            }
        }
    }