public TMasterResult handleReport(TReportRequest request) throws TException {
        TMasterResult result = new TMasterResult();
        TStatus tStatus = new TStatus(TStatusCode.OK);
        result.setStatus(tStatus);

        // get backend
        TBackend tBackend = request.getBackend();
        String host = tBackend.getHost();
        int bePort = tBackend.getBe_port();
        long beId;
        Backend backend = GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getBackendWithBePort(host, bePort);
        if (backend != null) {
            beId = backend.getId();
        } else {
            ComputeNode computeNode = null;
            // Compute node only reports resource usage or datacache metrics or tasks.
            if (request.isSetResource_usage() || request.isSetDatacache_metrics() || request.isSetTasks()) {
                computeNode =
                        GlobalStateMgr.getCurrentState().getNodeMgr().getClusterInfo().getComputeNodeWithBePort(host, bePort);
            }

            if (computeNode != null) {
                beId = computeNode.getId();
            } else {
                tStatus.setStatus_code(TStatusCode.INTERNAL_ERROR);
                List<String> errorMsgs = Lists.newArrayList();
                String accessibleHostPort = NetUtils.getHostPortInAccessibleFormat(host, bePort);
                errorMsgs.add("backend or compute node [" + accessibleHostPort + "] does not exist.");
                tStatus.setError_msgs(errorMsgs);
                return result;
            }
        }

        Map<TTaskType, Set<Long>> tasks = null;
        Map<String, TDisk> disks = null;
        Map<Long, TTablet> tablets = null;
        List<TWorkGroup> activeWorkGroups = null;
        TResourceUsage resourceUsage = null;
        TDataCacheMetrics dataCacheMetrics = null;
        long reportVersion = -1;

        ReportType reportType = ReportType.UNKNOWN_REPORT;
        if (request.isSetTasks()) {
            tasks = request.getTasks();
            reportType = ReportType.TASK_REPORT;
        }

        if (request.isSetDisks()) {
            if (reportType != ReportType.UNKNOWN_REPORT) {
                buildErrorResult(tStatus,
                        "invalid report request, multi fields " + reportType + " " + ReportType.TASK_REPORT);
                return result;
            }
            disks = request.getDisks();
            reportType = ReportType.DISK_REPORT;
        }

        if (request.isSetTablets()) {
            if (reportType != ReportType.UNKNOWN_REPORT) {
                buildErrorResult(tStatus,
                        "invalid report request, multi fields " + reportType + " " + ReportType.TABLET_REPORT);
                return result;
            }
            tablets = request.getTablets();
            reportVersion = request.getReport_version();
            reportType = ReportType.TABLET_REPORT;
        } else if (request.isSetTablet_list()) {
            if (reportType != ReportType.UNKNOWN_REPORT) {
                buildErrorResult(tStatus,
                        "invalid report request, multi fields " + reportType + " " + ReportType.TABLET_REPORT);
                return result;
            }
            // the 'tablets' member will be deprecated in the future.
            tablets = buildTabletMap(request.getTablet_list());
            reportVersion = request.getReport_version();
            reportType = ReportType.TABLET_REPORT;
        }

        if (backend != null && request.isSetTablet_max_compaction_score()) {
            backend.setTabletMaxCompactionScore(request.getTablet_max_compaction_score());
        }

        if (request.isSetActive_workgroups()) {
            if (reportType != ReportType.UNKNOWN_REPORT) {
                buildErrorResult(tStatus,
                        "invalid report request, multi fields " + reportType + " " + ReportType.RESOURCE_GROUP_REPORT);
                return result;
            }
            activeWorkGroups = request.active_workgroups;
            reportType = ReportType.RESOURCE_GROUP_REPORT;
        }

        if (request.isSetResource_usage()) {
            if (reportType != ReportType.UNKNOWN_REPORT) {
                buildErrorResult(tStatus,
                        "invalid report request, multi fields " + reportType + " " + ReportType.RESOURCE_USAGE_REPORT);
                return result;
            }

            resourceUsage = request.getResource_usage();
            reportType = ReportType.RESOURCE_USAGE_REPORT;
        }

        if (request.isSetDatacache_metrics()) {
            if (reportType != ReportType.UNKNOWN_REPORT) {
                buildErrorResult(tStatus,
                        "invalid report request, multi fields " + reportType + " " +
                                ReportType.DATACACHE_METRICS_REPORT);
                return result;
            }

            dataCacheMetrics = request.getDatacache_metrics();
            reportType = ReportType.DATACACHE_METRICS_REPORT;
        }

        List<TWorkGroupOp> workGroupOps =
                GlobalStateMgr.getCurrentState().getResourceGroupMgr().getResourceGroupsNeedToDeliver(beId);
        result.setWorkgroup_ops(workGroupOps);

        ReportTask reportTask =
                new ReportTask(beId, reportType, tasks, disks, tablets, reportVersion, activeWorkGroups, resourceUsage,
                        dataCacheMetrics);
        try {
            putToQueue(reportTask);
        } catch (Exception e) {
            tStatus.setStatus_code(TStatusCode.INTERNAL_ERROR);
            List<String> errorMsgs = Lists.newArrayList();
            errorMsgs.add("failed to put report task to queue. queue size: " + reportQueue.size());
            errorMsgs.add("err: " + e.getMessage());
            tStatus.setError_msgs(errorMsgs);

            LOG.warn(errorMsgs);
            return result;
        }

        LOG.debug("report received from be/computeNode {}. type: {}, current queue size: {}",
                beId, reportType, reportQueue.size());
        return result;
    }
@Test
    public void testHandleReport() throws TException {
        Backend be = new Backend(10001, "host1", 8000);
        ComputeNode cn = new ComputeNode(10002, "host2", 8000);

        new MockUp<SystemInfoService>() {
            @Mock
            public Backend getBackendWithBePort(String host, int bePort) {
                if (host.equals(be.getHost()) && bePort == be.getBePort()) {
                    return be;
                }
                return null;
            }

            @Mock
            public ComputeNode getComputeNodeWithBePort(String host, int bePort) {
                if (host.equals(cn.getHost()) && bePort == cn.getBePort()) {
                    return cn;
                }
                return null;
            }
        };

        ReportHandler handler = new ReportHandler();
        TResourceUsage resourceUsage = genResourceUsage(1, 2L, 3L, 100);

        {

            TReportRequest req = new TReportRequest();
            req.setResource_usage(resourceUsage);

            TBackend tcn = new TBackend();
            tcn.setHost(cn.getHost());
            tcn.setBe_port(cn.getBePort());
            req.setBackend(tcn);

            TMasterResult res = handler.handleReport(req);
            Assert.assertEquals(TStatusCode.OK, res.getStatus().getStatus_code());
        }

        {

            TReportRequest req = new TReportRequest();
            req.setResource_usage(resourceUsage);

            TBackend tbe = new TBackend();
            tbe.setHost(be.getHost());
            tbe.setBe_port(be.getBePort());
            req.setBackend(tbe);

            TMasterResult res = handler.handleReport(req);
            Assert.assertEquals(TStatusCode.OK, res.getStatus().getStatus_code());
        }

        {

            TReportRequest req = new TReportRequest();

            TBackend tcn = new TBackend();
            tcn.setHost(cn.getHost() + "NotExist");
            tcn.setBe_port(cn.getBePort());
            req.setBackend(tcn);

            TMasterResult res = handler.handleReport(req);
            Assert.assertEquals(TStatusCode.INTERNAL_ERROR, res.getStatus().getStatus_code());
        }
    }