NamedCountersAccumulableParam implements AccumulableParam<NamedCounters, Tuple2<String,Long>> { @Override public NamedCounters addInPlace(NamedCounters counters1, NamedCounters counters2) { for (String counterName2 : counters2.counterNames()) { counters1.increment(counterName2, counters2.currentValue(counterName2)); } return counters1; }  @Override NamedCounters addAccumulator(NamedCounters counters, Tuple2<String, Long> incrementValue); @Override NamedCounters addInPlace(NamedCounters counters1, NamedCounters counters2); @Override NamedCounters zero(NamedCounters initialCounters);  }
@Test public void addInPlace() { NamedCounters retNamedCounters = countersAccumulableParam.addInPlace(namedCounters1, namedCounters2); assertTrue(retNamedCounters == namedCounters1); assertThat(retNamedCounters.counterNames(), containsInAnyOrder(counterName1, counterName2, counterName3)); assertThat(retNamedCounters.currentValue(counterName1), is(3L)); assertThat(retNamedCounters.currentValue(counterName2), is(8L)); assertThat(retNamedCounters.currentValue(counterName3), is(5L)); }