public static Collection<PerStepNamespaceMetrics> convert(
      String stepName,
      Map<MetricName, Long> counters,
      Map<MetricName, LockFreeHistogram.Snapshot> histograms,
      Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedPerWorkerMetricsCache) {

    Map<String, PerStepNamespaceMetrics> metricsByNamespace = new HashMap<>();
    for (Entry<MetricName, Long> entry : counters.entrySet()) {
      MetricName metricName = entry.getKey();

      Optional<MetricValue> metricValue =
          convertCounterToMetricValue(metricName, entry.getValue(), parsedPerWorkerMetricsCache);
      if (!metricValue.isPresent()) {
        continue;
      }

      PerStepNamespaceMetrics stepNamespaceMetrics =
          metricsByNamespace.get(metricName.getNamespace());
      if (stepNamespaceMetrics == null) {
        stepNamespaceMetrics =
            new PerStepNamespaceMetrics()
                .setMetricValues(new ArrayList<>())
                .setOriginalStep(stepName)
                .setMetricsNamespace(metricName.getNamespace());
        metricsByNamespace.put(metricName.getNamespace(), stepNamespaceMetrics);
      }

      stepNamespaceMetrics.getMetricValues().add(metricValue.get());
    }

    for (Entry<MetricName, LockFreeHistogram.Snapshot> entry : histograms.entrySet()) {
      MetricName metricName = entry.getKey();
      Optional<MetricValue> metricValue =
          convertHistogramToMetricValue(metricName, entry.getValue(), parsedPerWorkerMetricsCache);
      if (!metricValue.isPresent()) {
        continue;
      }

      PerStepNamespaceMetrics stepNamespaceMetrics =
          metricsByNamespace.get(metricName.getNamespace());
      if (stepNamespaceMetrics == null) {
        stepNamespaceMetrics =
            new PerStepNamespaceMetrics()
                .setMetricValues(new ArrayList<>())
                .setOriginalStep(stepName)
                .setMetricsNamespace(metricName.getNamespace());
        metricsByNamespace.put(metricName.getNamespace(), stepNamespaceMetrics);
      }

      stepNamespaceMetrics.getMetricValues().add(metricValue.get());
    }

    return metricsByNamespace.values();
  }
@Test
  public void testConvert_convertCountersAndHistograms() {
    String step = "testStep";
    Map<MetricName, Long> counters = new HashMap<>();
    Map<MetricName, LockFreeHistogram.Snapshot> histograms = new HashMap<>();
    Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedMetricNames = new HashMap<>();

    MetricName counterMetricName = MetricName.named("BigQuerySink", "counter*label1:val1;");
    counters.put(counterMetricName, 3L);

    MetricName histogramMetricName = MetricName.named("BigQuerySink", "histogram*label2:val2;");
    LockFreeHistogram linearHistogram = new LockFreeHistogram(histogramMetricName, lienarBuckets);
    linearHistogram.update(5.0);
    histograms.put(histogramMetricName, linearHistogram.getSnapshotAndReset().get());

    Collection<PerStepNamespaceMetrics> conversionResult =
        MetricsToPerStepNamespaceMetricsConverter.convert(
            step, counters, histograms, parsedMetricNames);

    // Expected counter MetricValue
    Map<String, String> counterLabelMap = new HashMap<>();
    counterLabelMap.put("label1", "val1");
    MetricValue expectedCounter =
        new MetricValue().setMetric("counter").setValueInt64(3L).setMetricLabels(counterLabelMap);

    // Expected histogram MetricValue
    List<Long> bucketCounts1 = ImmutableList.of(1L);

    Linear linearOptions1 = new Linear().setNumberOfBuckets(10).setWidth(10.0).setStart(0.0);
    BucketOptions bucketOptions1 = new BucketOptions().setLinear(linearOptions1);

    DataflowHistogramValue linearHistogram1 =
        new DataflowHistogramValue()
            .setCount(1L)
            .setBucketOptions(bucketOptions1)
            .setBucketCounts(bucketCounts1);

    Map<String, String> histogramLabelMap = new HashMap<>();
    histogramLabelMap.put("label2", "val2");

    MetricValue expectedHistogram =
        new MetricValue()
            .setMetric("histogram")
            .setMetricLabels(histogramLabelMap)
            .setValueHistogram(linearHistogram1);

    assertThat(conversionResult.size(), equalTo(1));
    PerStepNamespaceMetrics perStepNamespaceMetrics = conversionResult.iterator().next();

    assertThat(perStepNamespaceMetrics.getOriginalStep(), equalTo(step));
    assertThat(perStepNamespaceMetrics.getMetricsNamespace(), equalTo("BigQuerySink"));
    assertThat(perStepNamespaceMetrics.getMetricValues().size(), equalTo(2));
    assertThat(
        perStepNamespaceMetrics.getMetricValues(),
        containsInAnyOrder(expectedCounter, expectedHistogram));

    // Verify that parsedMetricNames have been cached.
    LabeledMetricNameUtils.ParsedMetricName parsedCounterMetricName =
        LabeledMetricNameUtils.parseMetricName(counterMetricName.getName()).get();
    LabeledMetricNameUtils.ParsedMetricName parsedHistogramMetricName =
        LabeledMetricNameUtils.parseMetricName(histogramMetricName.getName()).get();

    assertThat(parsedMetricNames.size(), equalTo(2));
    assertThat(
        parsedMetricNames, IsMapContaining.hasEntry(counterMetricName, parsedCounterMetricName));
    assertThat(
        parsedMetricNames,
        IsMapContaining.hasEntry(histogramMetricName, parsedHistogramMetricName));
  }