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

    Map<MetricName, Long> counters = new HashMap<>();
    MetricName invalidName1 = MetricName.named("BigQuerySink", "**");
    counters.put(invalidName1, 5L);

    Map<MetricName, LockFreeHistogram.Snapshot> histograms = new HashMap<>();
    MetricName invalidName2 = MetricName.named("BigQuerySink", "****");
    LockFreeHistogram nonEmptyLinearHistogram = new LockFreeHistogram(invalidName2, lienarBuckets);
    nonEmptyLinearHistogram.update(-5.0);
    histograms.put(invalidName2, nonEmptyLinearHistogram.getSnapshotAndReset().get());

    Collection<PerStepNamespaceMetrics> conversionResult =
        MetricsToPerStepNamespaceMetricsConverter.convert(
            "testStep", counters, histograms, parsedMetricNames);
    assertThat(conversionResult.size(), equalTo(0));
    assertThat(parsedMetricNames.size(), equalTo(0));
  }