public void commitLogical(CommittedBundle<?> bundle, MetricUpdates updates) {
    for (MetricUpdate<Long> counter : updates.counterUpdates()) {
      counters.get(counter.getKey()).commitLogical(bundle, counter.getUpdate());
    }
    for (MetricUpdate<DistributionData> distribution : updates.distributionUpdates()) {
      distributions.get(distribution.getKey()).commitLogical(bundle, distribution.getUpdate());
    }
    for (MetricUpdate<GaugeData> gauge : updates.gaugeUpdates()) {
      gauges.get(gauge.getKey()).commitLogical(bundle, gauge.getUpdate());
    }
    for (MetricUpdate<StringSetData> sSet : updates.stringSetUpdates()) {
      stringSet.get(sSet.getKey()).commitLogical(bundle, sSet.getUpdate());
    }
  }
@SuppressWarnings("unchecked")
  @Test
  public void testApplyCommittedNoFilter() {
    metrics.commitLogical(
        bundle1,
        MetricUpdates.create(
            ImmutableList.of(
                MetricUpdate.create(MetricKey.create("step1", NAME1), 5L),
                MetricUpdate.create(MetricKey.create("step1", NAME2), 8L)),
            ImmutableList.of(
                MetricUpdate.create(
                    MetricKey.create("step1", NAME1), DistributionData.create(8, 2, 3, 5))),
            ImmutableList.of(
                MetricUpdate.create(MetricKey.create("step1", NAME4), GaugeData.create(15L))),
            ImmutableList.of(
                MetricUpdate.create(
                    MetricKey.create("step1", NAME4),
                    StringSetData.create(ImmutableSet.of("ab"))))));
    metrics.commitLogical(
        bundle1,
        MetricUpdates.create(
            ImmutableList.of(
                MetricUpdate.create(MetricKey.create("step2", NAME1), 7L),
                MetricUpdate.create(MetricKey.create("step1", NAME2), 4L)),
            ImmutableList.of(
                MetricUpdate.create(
                    MetricKey.create("step1", NAME1), DistributionData.create(4, 1, 4, 4))),
            ImmutableList.of(
                MetricUpdate.create(MetricKey.create("step1", NAME4), GaugeData.create(27L))),
            ImmutableList.of(
                MetricUpdate.create(
                    MetricKey.create("step1", NAME4),
                    StringSetData.create(ImmutableSet.of("cd"))))));

    MetricQueryResults results = metrics.allMetrics();
    assertThat(
        results.getCounters(),
        containsInAnyOrder(
            attemptedMetricsResult("ns1", "name1", "step1", 0L),
            attemptedMetricsResult("ns1", "name2", "step1", 0L),
            attemptedMetricsResult("ns1", "name1", "step2", 0L)));
    assertThat(
        results.getCounters(),
        containsInAnyOrder(
            committedMetricsResult("ns1", "name1", "step1", 5L),
            committedMetricsResult("ns1", "name2", "step1", 12L),
            committedMetricsResult("ns1", "name1", "step2", 7L)));
    assertThat(
        results.getDistributions(),
        contains(
            attemptedMetricsResult("ns1", "name1", "step1", DistributionResult.IDENTITY_ELEMENT)));
    assertThat(
        results.getDistributions(),
        contains(
            committedMetricsResult(
                "ns1", "name1", "step1", DistributionResult.create(12, 3, 3, 5))));
    assertThat(
        results.getGauges(),
        contains(attemptedMetricsResult("ns2", "name2", "step1", GaugeResult.empty())));
    assertThat(
        results.getGauges(),
        contains(
            committedMetricsResult(
                "ns2", "name2", "step1", GaugeResult.create(27L, Instant.now()))));
    assertThat(
        results.getStringSets(),
        contains(
            committedMetricsResult(
                "ns2", "name2", "step1", StringSetResult.create(ImmutableSet.of("ab", "cd")))));
  }