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_successfulyConvertCounters() {
    String step = "testStepName";
    Map<MetricName, LockFreeHistogram.Snapshot> emptyHistograms = new HashMap<>();
    Map<MetricName, Long> counters = new HashMap<MetricName, Long>();
    Map<MetricName, LabeledMetricNameUtils.ParsedMetricName> parsedMetricNames = new HashMap<>();

    MetricName bigQueryMetric1 = MetricName.named("BigQuerySink", "metric1");
    MetricName bigQueryMetric2 =
        MetricName.named("BigQuerySink", "metric2*label1:val1;label2:val2;");
    MetricName bigQueryMetric3 = MetricName.named("BigQuerySink", "zeroValue");

    counters.put(bigQueryMetric1, 5L);
    counters.put(bigQueryMetric2, 10L);
    counters.put(bigQueryMetric3, 0L);

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

    MetricValue expectedVal1 =
        new MetricValue().setMetric("metric1").setValueInt64(5L).setMetricLabels(new HashMap<>());
    Map<String, String> val2LabelMap = new HashMap<>();
    val2LabelMap.put("label1", "val1");
    val2LabelMap.put("label2", "val2");
    MetricValue expectedVal2 =
        new MetricValue().setMetric("metric2").setValueInt64(10L).setMetricLabels(val2LabelMap);

    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));

    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));
  }