@Override
  /**
   * {@inheritDoc} Handles the bundle's completion report. Parses the monitoringInfos in the
   * response, then updates the MetricsRegistry.
   */
  public void onCompleted(BeamFnApi.ProcessBundleResponse response) {
    response.getMonitoringInfosList().stream()
        .filter(monitoringInfo -> !monitoringInfo.getPayload().isEmpty())
        .map(this::parseAndUpdateMetric)
        .distinct()
        .forEach(samzaMetricsContainer::updateMetrics);
  }
@Test
  public void testCounter() {
    // Hex for 123
    byte[] payload = "\173".getBytes(Charset.defaultCharset());

    MetricsApi.MonitoringInfo monitoringInfo =
        MetricsApi.MonitoringInfo.newBuilder()
            .setType(SUM_INT64_TYPE)
            .setPayload(ByteString.copyFrom(payload))
            .putLabels(MonitoringInfoConstants.Labels.NAMESPACE, EXPECTED_NAMESPACE)
            .putLabels(MonitoringInfoConstants.Labels.NAME, EXPECTED_COUNTER_NAME)
            .build();
    BeamFnApi.ProcessBundleResponse response =
        BeamFnApi.ProcessBundleResponse.newBuilder().addMonitoringInfos(monitoringInfo).build();

    // Execute
    samzaMetricsBundleProgressHandler.onCompleted(response);

    // Verify
    MetricName metricName = MetricName.named(EXPECTED_NAMESPACE, EXPECTED_COUNTER_NAME);
    CounterCell counter =
        (CounterCell) samzaMetricsContainer.getContainer(stepName).getCounter(metricName);

    assertEquals(counter.getCumulative(), (Long) 123L);
  }