@Test public void testMasterService() throws Exception {

  try {
    LOG.info("===========================testMasterService===============================");
    Worker worker = LocalClusterContext.get().getWorker(worker0Attempt0Id).getWorker();
    Location masterLoc =
      LocalClusterContext.get().getMaster().getAppMaster().getAppContext().getMasterService()
        .getLocation();
    TConnection connection = TConnectionManager.getConnection(worker.getConf());
    MasterProtocol master = connection.getMasterService(masterLoc.getIp(), masterLoc.getPort());

    int psAgentId = master
      .getPSAgentId(null, PSAgentMasterServiceProtos.GetPSAgentIdRequest.getDefaultInstance())
      .getPsAgentId();

    // worker register
    WorkerAttemptId worker1Attempt0Id =
      new WorkerAttemptId(new WorkerId(new WorkerGroupId(1), 0), 0);
    WorkerRegisterRequest registeRequest =
      WorkerRegisterRequest.newBuilder().setPsAgentId(psAgentId)
        .setWorkerAttemptId(ProtobufUtil.convertToIdProto(worker1Attempt0Id))
        .setLocation(LocationProto.newBuilder().setIp("10.10.10.10").setPort(10000).build())
        .build();
    WorkerRegisterResponse registerResponse = master.workerRegister(null, registeRequest);
    assertTrue(registerResponse.getCommand() == WorkerCommandProto.W_SHUTDOWN);

    WorkerReportRequest.Builder reportBuilder = WorkerReportRequest.newBuilder();
    Pair.Builder kvBuilder = Pair.newBuilder();
    TaskStateProto.Builder taskBuilder = TaskStateProto.newBuilder();

    reportBuilder.setWorkerAttemptId(ProtobufUtil.convertToIdProto(worker0Attempt0Id));

    taskBuilder.setProgress(0.20f);
    taskBuilder.setState("RUNNING");
    taskBuilder.setTaskId(ProtobufUtil.convertToIdProto(task0Id));
    kvBuilder.setKey("task_key1");
    kvBuilder.setValue("100");
    taskBuilder.addCounters(kvBuilder.build());
    kvBuilder.setKey("task_key2");
    kvBuilder.setValue("200");
    taskBuilder.addCounters(kvBuilder.build());
    reportBuilder.addTaskReports(taskBuilder.build());

    taskBuilder.setProgress(0.30f);
    taskBuilder.setState("RUNNING");
    taskBuilder.setTaskId(ProtobufUtil.convertToIdProto(task1Id));
    kvBuilder.setKey("task_key1");
    kvBuilder.setValue("1000");
    taskBuilder.addCounters(kvBuilder.build());
    kvBuilder.setKey("task_key2");
    kvBuilder.setValue("2000");
    taskBuilder.addCounters(kvBuilder.build());
    reportBuilder.addTaskReports(taskBuilder.build());

    kvBuilder.setKey("worker_key1");
    kvBuilder.setValue("100");
    reportBuilder.addPairs(kvBuilder.build());
    kvBuilder.setKey("worker_key2");
    kvBuilder.setValue("200");
    reportBuilder.addPairs(kvBuilder.build());

    WorkerReportResponse reportResponse = master.workerReport(null, reportBuilder.build());
    assertTrue(reportResponse.getCommand() == WorkerCommandProto.W_SUCCESS);
    assertEquals(reportResponse.getActiveTaskNum(), 2);

    AngelApplicationMaster angelAppMaster = LocalClusterContext.get().getMaster().getAppMaster();
    WorkerAttempt worker0Attempt =
      angelAppMaster.getAppContext().getWorkerManager().getWorker(worker0Attempt0Id.getWorkerId())
        .getWorkerAttempt(worker0Attempt0Id);
    assertTrue(worker0Attempt != null);
    Map<String, String> workerMetrics = worker0Attempt.getMetrics();
    String valueForWorkerKey1 = workerMetrics.get("worker_key1");
    String valueForWorkerKey2 = workerMetrics.get("worker_key2");
    assertNotNull(valueForWorkerKey1);
    assertNotNull(valueForWorkerKey2);
    assertEquals(valueForWorkerKey1, "100");
    assertEquals(valueForWorkerKey2, "200");

    AMTaskManager amTaskManager = angelAppMaster.getAppContext().getTaskManager();
    AMTask task0 = amTaskManager.getTask(task0Id);
    AMTask task1 = amTaskManager.getTask(task1Id);
    assertTrue(task0 != null);
    assertTrue(task1 != null);
    Map<String, String> task0Metrics = task0.getMetrics();
    Map<String, String> task1Metrics = task1.getMetrics();
    String valueForTask0Key1 = task0Metrics.get("task_key1");
    String valueForTask0Key2 = task0Metrics.get("task_key2");
    String valueForTask1Key1 = task1Metrics.get("task_key1");
    String valueForTask1Key2 = task1Metrics.get("task_key2");
    assertTrue(valueForTask0Key1 != null);
    assertTrue(valueForTask0Key2 != null);
    assertTrue(valueForTask1Key1 != null);
    assertTrue(valueForTask1Key2 != null);
    assertEquals(valueForTask0Key1, "100");
    assertEquals(valueForTask0Key2, "200");
    assertEquals(valueForTask1Key1, "1000");
    assertEquals(valueForTask1Key2, "2000");
    assertEquals(task0.getProgress(), 0.20f, 0.000001);
    assertEquals(task1.getProgress(), 0.30f, 0.000001);
  } catch (Exception x) {
    LOG.error("run testMasterService failed ", x);
    throw x;
  }
}