public static Counter throttledTimeCounter(RpcMethod method) {

    LabeledMetricNameUtils.MetricNameBuilder nameBuilder =
        LabeledMetricNameUtils.MetricNameBuilder.baseNameBuilder(THROTTLED_TIME);
    nameBuilder.addLabel(RPC_METHOD, method.toString());
    MetricName metricName = nameBuilder.build(METRICS_NAMESPACE);
    // for specific method
    Counter fineCounter = new DelegatingCounter(metricName, false, true);
    // for overall throttling time, used by runner for scaling decision
    Counter coarseCounter = BigQueryServicesImpl.StorageClientImpl.THROTTLING_MSECS;
    return new NestedCounter(
        MetricName.named(
            METRICS_NAMESPACE, metricName.getName() + coarseCounter.getName().getName()),
        fineCounter,
        coarseCounter);
  }
@Test
  public void testThrottledTimeCounter() throws Exception {
    // Setup
    TestMetricsContainer testContainer = new TestMetricsContainer();
    MetricsEnvironment.setCurrentContainer(testContainer);

    // Test throttleCounter metric.
    Counter appendRowsThrottleCounter =
        BigQuerySinkMetrics.throttledTimeCounter(BigQuerySinkMetrics.RpcMethod.APPEND_ROWS);
    appendRowsThrottleCounter.inc(1);
    assertThat(
        appendRowsThrottleCounter.getName().getName(),
        equalTo("ThrottledTime*rpc_method:APPEND_ROWS;throttling-msecs"));

    // check that both sub-counters have been incremented
    MetricName counterName =
        MetricName.named("BigQuerySink", "ThrottledTime*rpc_method:APPEND_ROWS;");
    testContainer.assertPerWorkerCounterValue(counterName, 1L);

    counterName =
        MetricName.named(
            BigQueryServicesImpl.StorageClientImpl.class, Metrics.THROTTLE_TIME_COUNTER_NAME);
    assertEquals(1L, (long) testContainer.getCounter(counterName).getCumulative());
  }