public synchronized Command heartbeat(final long workerId,
      final Map<String, Long> capacityBytesOnTiers, final Map<String, Long> usedBytesOnTiers,
      final List<Long> removedBlocks, final Map<BlockStoreLocation, List<Long>> addedBlocks,
      final Map<String, List<String>> lostStorage, final List<Metric> metrics)
      throws IOException {
    final BlockHeartbeatPOptions options = BlockHeartbeatPOptions.newBuilder()
        .addAllMetrics(metrics).putAllCapacityBytesOnTiers(capacityBytesOnTiers).build();

    final List<LocationBlockIdListEntry> entryList = convertBlockListMapToProto(addedBlocks);

    final Map<String, StorageList> lostStorageMap = lostStorage.entrySet().stream()
        .collect(Collectors.toMap(Map.Entry::getKey,
            e -> StorageList.newBuilder().addAllStorage(e.getValue()).build()));

    final BlockHeartbeatPRequest request = BlockHeartbeatPRequest.newBuilder().setWorkerId(workerId)
        .putAllUsedBytesOnTiers(usedBytesOnTiers).addAllRemovedBlockIds(removedBlocks)
        .addAllAddedBlocks(entryList).setOptions(options)
        .putAllLostStorage(lostStorageMap).build();

    return retryRPC(() -> mClient.withDeadlineAfter(mContext.getClusterConf()
        .getMs(PropertyKey.WORKER_MASTER_PERIODICAL_RPC_TIMEOUT), TimeUnit.MILLISECONDS)
        .blockHeartbeat(request).getCommand(), LOG, "Heartbeat", "workerId=%d", workerId);
  }
@Test
  public void heartBeat() throws Exception {
    final long workerId = 1L;
    final Map<String, Long> capacityBytesOnTiers = ImmutableMap.of("MEM", 1024 * 1024L);
    final Map<String, Long> usedBytesOnTiers = ImmutableMap.of("MEM", 1024L);
    final List<Long> removedBlocks = ImmutableList.of();
    final Map<BlockStoreLocation, List<Long>> addedBlocks = ImmutableMap.of(
        new BlockStoreLocation("MEM", 0, "MEM"),
        ImmutableList.of(11L, 12L, 13L)
    );
    final Map<String, List<String>> lostStorage = ImmutableMap.of(
        "MEM",
        ImmutableList.of("/tmp/lost")
    );
    final List<Metric> metrics = ImmutableList.of();

    createMockService(
        new BlockMasterWorkerServiceGrpc.BlockMasterWorkerServiceImplBase() {
          @Override
          public void blockHeartbeat(BlockHeartbeatPRequest request,
                                     StreamObserver<BlockHeartbeatPResponse> responseObserver) {

            // verify request data
            assertEquals(workerId, request.getWorkerId());
            assertEquals(usedBytesOnTiers, request.getUsedBytesOnTiersMap());
            assertEquals(removedBlocks, request.getRemovedBlockIdsList());

            // verify added blocks
            for (LocationBlockIdListEntry entry: request.getAddedBlocksList()) {
              BlockStoreLocationProto locationProto = entry.getKey();
              BlockStoreLocation location = new BlockStoreLocation(
                  locationProto.getTierAlias(),
                  0,
                  locationProto.getMediumType()
              );
              List<Long> blocks = addedBlocks.get(location);
              assertTrue(
                  blocks != null && blocks.containsAll(entry.getValue().getBlockIdList())
              );
            }

            // verify lost storage
            assertEquals(lostStorage.entrySet().stream()
                    .collect(Collectors.toMap(Map.Entry::getKey,
                        e -> StorageList.newBuilder().addAllStorage(e.getValue()).build())),
                request.getLostStorageMap());

            responseObserver.onNext(
                BlockHeartbeatPResponse.newBuilder().setCommand(Command.newBuilder()
                .setCommandType(
                    CommandType.Nothing)).build()
            );
            responseObserver.onCompleted();
          }
        });

    BlockMasterClient client = new BlockMasterClient(
        MasterClientContext.newBuilder(ClientContext.create(mConf)).build()
    );

    assertEquals(CommandType.Nothing, client.heartbeat(
        workerId,
        capacityBytesOnTiers,
        usedBytesOnTiers,
        removedBlocks,
        addedBlocks,
        lostStorage,
        metrics).getCommandType());
  }