DistributionCounterUpdateAggregator implements CounterUpdateAggregator { @Override public CounterUpdate aggregate(List<CounterUpdate> counterUpdates) { if (counterUpdates == null || counterUpdates.isEmpty()) { return null; } if (counterUpdates.stream().anyMatch(c -> c.getDistribution() == null)) { throw new UnsupportedOperationException( "Aggregating DISTRIBUTION counter updates over non-distribution type is not implemented."); } CounterUpdate initial = counterUpdates.remove(0); return counterUpdates.stream() .reduce( initial, (first, second) -> first.setDistribution( new DistributionUpdate() .setCount( longToSplitInt( splitIntToLong(first.getDistribution().getCount()) + splitIntToLong(second.getDistribution().getCount()))) .setMax( longToSplitInt( Math.max( splitIntToLong(first.getDistribution().getMax()), splitIntToLong(second.getDistribution().getMax())))) .setMin( longToSplitInt( Math.min( splitIntToLong(first.getDistribution().getMin()), splitIntToLong(second.getDistribution().getMin())))) .setSum( longToSplitInt( splitIntToLong(first.getDistribution().getSum()) + splitIntToLong(second.getDistribution().getSum()))))); }  @Override CounterUpdate aggregate(List<CounterUpdate> counterUpdates);  }
@Test public void testAggregate() { CounterUpdate combined = aggregator.aggregate(counterUpdates); assertEquals(45L, splitIntToLong(combined.getDistribution().getSum())); assertEquals(10L, splitIntToLong(combined.getDistribution().getCount())); assertEquals(9L, splitIntToLong(combined.getDistribution().getMax())); assertEquals(0L, splitIntToLong(combined.getDistribution().getMin())); }
@Test(expected = UnsupportedOperationException.class) public void testAggregateWithNullIntegerDistribution() { counterUpdates.get(0).setDistribution(null); aggregator.aggregate(counterUpdates); }