code: public synchronized Command heartbeat(final long workerId,
        final Map<String, Long> capacityBytesOnTiers, final Map<String, Long> usedBytesOnTiers,
        final List<Long> removedBlocks, final Map<String, 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 Map<String, TierList> addedBlocksMap = addedBlocks.entrySet().stream()
          .collect(Collectors.toMap(Map.Entry::getKey,
              e -> TierList.newBuilder().addAllTiers(e.getValue()).build()));
  
      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)
          .putAllAddedBlocksOnTiers(addedBlocksMap).setOptions(options)
          .putAllLostStorage(lostStorageMap).build();
  
      return retryRPC(() -> mClient.blockHeartbeat(request).getCommand());
    }
docs: The method the worker should periodically execute to heartbeat back to the master.
  
  @param workerId the worker id
  @param capacityBytesOnTiers a mapping from storage tier alias to capacity bytes
  @param usedBytesOnTiers a mapping from storage tier alias to used bytes
  @param removedBlocks a list of block removed from this worker
  @param addedBlocks a mapping from storage tier alias to added blocks
  @param lostStorage a mapping from storage tier alias to a list of lost storage paths
  @param metrics a list of worker metrics
  @return an optional command for the worker to execute