public static Counter appendRowsRowStatusCounter(
      RowStatus rowStatus, String rpcStatus, String tableId) {
    LabeledMetricNameUtils.MetricNameBuilder nameBuilder =
        LabeledMetricNameUtils.MetricNameBuilder.baseNameBuilder(APPEND_ROWS_ROW_STATUS);
    nameBuilder.addLabel(ROW_STATUS, rowStatus.toString());
    nameBuilder.addLabel(RPC_STATUS_LABEL, rpcStatus);
    if (BigQuerySinkMetrics.supportMetricsDeletion) {
      nameBuilder.addLabel(TABLE_ID_LABEL, tableId);
    }

    MetricName metricName = nameBuilder.build(METRICS_NAMESPACE);

    return new DelegatingCounter(metricName, false, true);
  }
@Test
  public void testAppendRowsRowStatusCounter() throws Exception {
    // Setup
    TestMetricsContainer testContainer = new TestMetricsContainer();
    MetricsEnvironment.setCurrentContainer(testContainer);

    BigQuerySinkMetrics.setSupportMetricsDeletion(false);
    Counter deletesDisabledCounter =
        BigQuerySinkMetrics.appendRowsRowStatusCounter(
            BigQuerySinkMetrics.RowStatus.SUCCESSFUL, "rpcStatus", "tableId");
    deletesDisabledCounter.inc();
    MetricName deletesDisabledCounterName =
        MetricName.named(
            "BigQuerySink", "RowsAppendedCount*row_status:SUCCESSFUL;rpc_status:rpcStatus;");
    testContainer.assertPerWorkerCounterValue(deletesDisabledCounterName, 1L);

    BigQuerySinkMetrics.setSupportMetricsDeletion(true);
    testContainer.reset();
    Counter deletesEnabledCounter =
        BigQuerySinkMetrics.appendRowsRowStatusCounter(
            BigQuerySinkMetrics.RowStatus.SUCCESSFUL, "rpcStatus", "tableId");
    deletesEnabledCounter.inc();
    MetricName deletesEnabledCounterName =
        MetricName.named(
            "BigQuerySink",
            "RowsAppendedCount*row_status:SUCCESSFUL;rpc_status:rpcStatus;table_id:tableId;");
    testContainer.assertPerWorkerCounterValue(deletesEnabledCounterName, 1L);
  }