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_successfulConvertHistograms() {
    Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedMetricNames = new HashMap<>();

    Map<MetricName, LockFreeHistogram.Snapshot> histograms = new HashMap<>();
    MetricName bigQueryMetric1 = MetricName.named("BigQuerySink", "baseLabel");
    MetricName bigQueryMetric2 =
        MetricName.named("BigQuerySink", "baseLabel*label1:val1;label2:val2;");
    MetricName bigQueryMetric3 = MetricName.named("BigQuerySink", "zeroValue");

    LockFreeHistogram nonEmptyLinearHistogram =
        new LockFreeHistogram(bigQueryMetric1, lienarBuckets);
    nonEmptyLinearHistogram.update(-5.0, 15.0, 25.0, 35.0, 105.0);
    histograms.put(bigQueryMetric1, nonEmptyLinearHistogram.getSnapshotAndReset().get());

    LockFreeHistogram noEmptyExponentialHistogram =
        new LockFreeHistogram(bigQueryMetric2, exponentialBuckets);
    noEmptyExponentialHistogram.update(-5.0, 15.0, 25.0, 35.0, 105.0);
    histograms.put(bigQueryMetric2, noEmptyExponentialHistogram.getSnapshotAndReset().get());

    LockFreeHistogram.Snapshot emptySnapshot =
        LockFreeHistogram.Snapshot.create(
            LockFreeHistogram.OutlierStatistic.EMPTY,
            LockFreeHistogram.OutlierStatistic.EMPTY,
            ImmutableLongArray.of(),
            lienarBuckets);
    histograms.put(bigQueryMetric3, emptySnapshot);

    String step = "testStep";
    Map<MetricName, Long> emptyCounters = new HashMap<>();
    Collection<PerStepNamespaceMetrics> conversionResult =
        MetricsToPerStepNamespaceMetricsConverter.convert(
            step, emptyCounters, histograms, parsedMetricNames);

    // Expected value 1
    List<Long> bucketCounts1 = ImmutableList.of(0L, 1L, 1L, 1L);

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

    OutlierStats outlierStats1 =
        new OutlierStats()
            .setUnderflowCount(1L)
            .setUnderflowMean(-5.0)
            .setOverflowCount(1L)
            .setOverflowMean(105.0);
    DataflowHistogramValue linearHistogram1 =
        new DataflowHistogramValue()
            .setCount(5L)
            .setBucketOptions(bucketOptions1)
            .setBucketCounts(bucketCounts1)
            .setOutlierStats(outlierStats1);

    MetricValue expectedVal1 =
        new MetricValue()
            .setMetric("baseLabel")
            .setMetricLabels(new HashMap<>())
            .setValueHistogram(linearHistogram1);

    // Expected value 2
    List<Long> bucketCounts2 = ImmutableList.of(0L, 0L, 0L, 1L, 1L);
    OutlierStats outlierStats2 =
        new OutlierStats()
            .setUnderflowCount(1L)
            .setUnderflowMean(-5.0)
            .setOverflowCount(2L)
            .setOverflowMean(70.0);
    Base2Exponent exponentialOptions2 = new Base2Exponent().setNumberOfBuckets(5).setScale(0);

    BucketOptions bucketOptions2 = new BucketOptions().setExponential(exponentialOptions2);

    DataflowHistogramValue exponentialHistogram2 =
        new DataflowHistogramValue()
            .setCount(5L)
            .setBucketOptions(bucketOptions2)
            .setBucketCounts(bucketCounts2)
            .setOutlierStats(outlierStats2);

    Map<String, String> metric2Labels = new HashMap<>();
    metric2Labels.put("label1", "val1");
    metric2Labels.put("label2", "val2");
    MetricValue expectedVal2 =
        new MetricValue()
            .setMetric("baseLabel")
            .setValueHistogram(exponentialHistogram2)
            .setMetricLabels(metric2Labels);

    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(expectedVal1, expectedVal2));

    // Verify that parsedMetricNames have been cached.
    LabeledMetricNameUtils.ParsedMetricName parsedBigQueryMetric1 =
        LabeledMetricNameUtils.parseMetricName(bigQueryMetric1.getName()).get();
    LabeledMetricNameUtils.ParsedMetricName parsedBigQueryMetric2 =
        LabeledMetricNameUtils.parseMetricName(bigQueryMetric2.getName()).get();

    assertThat(parsedMetricNames.size(), equalTo(2));
    assertThat(parsedMetricNames, IsMapContaining.hasEntry(bigQueryMetric1, parsedBigQueryMetric1));
    assertThat(parsedMetricNames, IsMapContaining.hasEntry(bigQueryMetric2, parsedBigQueryMetric2));
  }