class BoundedTimestampSetAggregator extends KorypheBinaryOperator<BoundedTimestampSet> {

    @Override
    protected BoundedTimestampSet _apply(final BoundedTimestampSet a, final BoundedTimestampSet b) {
        // Can only be merged if they have the same time bucket and maximum size.
        if (!a.getTimeBucket().equals(b.getTimeBucket())) {
            throw new IllegalArgumentException("Can only merge two BoundedTimestampSet with the same time bucket:" +
                    "a has time bucket " + a.getTimeBucket() + ", b has time bucket " + b.getTimeBucket());
        }
        if (a.getMaxSize() != b.getMaxSize()) {
            throw new IllegalArgumentException("Can only merge two BoundedTimestampSet with the same maximum size:" +
                    "a has maximum size " + a.getMaxSize() + ", b has maximum size " + b.getMaxSize());
        }
        // If both are in NOT_FULL state then all the timestamps for b can just be added to a.
        if (a.getState().equals(BoundedTimestampSet.State.NOT_FULL)
                && b.getState().equals(BoundedTimestampSet.State.NOT_FULL)) {
            a.add(b.getTimestamps());
            return a;
        }
        // If either a or b or both are in SAMPLE state, then the aggregated version must also be in SAMPLE state.
        if (a.getState().equals(BoundedTimestampSet.State.NOT_FULL)) {
            // Need to switch the state of a to SAMPLE and add in b's samples.
            a.switchToSampleState();
            a.add(b.getTimestamps());
        } else {
            a.add(b.getTimestamps());
        }
        return a;
    }

}