SumCounterUpdateAggregator implements CounterUpdateAggregator { @Override public CounterUpdate aggregate(List<CounterUpdate> counterUpdates) { if (counterUpdates == null || counterUpdates.isEmpty()) { return null; } if (counterUpdates.stream().anyMatch(c -> c.getInteger() == null)) { throw new UnsupportedOperationException( "Aggregating SUM counter updates over non-integer type is not implemented."); } CounterUpdate initial = counterUpdates.remove(0); return counterUpdates.stream() .reduce( initial, (first, second) -> first.setInteger( longToSplitInt( splitIntToLong(first.getInteger()) + splitIntToLong(second.getInteger())))); }  @Override CounterUpdate aggregate(List<CounterUpdate> counterUpdates);  }
@Test public void testAggregate() { CounterUpdate combined = aggregator.aggregate(counterUpdates); assertEquals(45L, splitIntToLong(combined.getInteger())); }
@Test(expected = UnsupportedOperationException.class) public void testAggregateWithNullInteger() { counterUpdates.get(0).setInteger(null); aggregator.aggregate(counterUpdates); }