  @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;
    }
  }