public void putWorkerMetrics(String source, List<Metric> metrics) {
    if (metrics.isEmpty() || source == null) {
      return;
    }
    try (LockResource r = new LockResource(mLock.readLock())) {
      putReportedMetrics(InstanceType.WORKER, metrics);
    }
    LOG.debug("Put {} metrics of worker {}", metrics.size(), source);
  }
@Test
  public void putWorkerMetrics() {
    String workerHost1 = "192_1_1_1";
    List<Metric> metrics1 = Lists.newArrayList(
        Metric.from(MetricKey.WORKER_BYTES_READ_REMOTE.getName() + "." + workerHost1,
            10, MetricType.COUNTER),
        Metric.from(MetricKey.WORKER_BYTES_READ_DOMAIN.getName() + "." + workerHost1,
            20, MetricType.COUNTER));
    mMetricStore.putWorkerMetrics(workerHost1, metrics1);

    String workerHost2 = "192_1_1_2";
    List<Metric> metrics2 = Lists.newArrayList(
        Metric.from(MetricKey.WORKER_BYTES_READ_REMOTE.getName() + "." + workerHost2,
            1, MetricType.COUNTER));
    mMetricStore.putWorkerMetrics(workerHost2, metrics2);
    assertEquals(11,
        MetricsSystem.counter(MetricKey.CLUSTER_BYTES_READ_REMOTE.getName()).getCount());
  }