BatchingAggregator.java

public class BatchingAggregator<T> {
    private final int maxBatchSize;
    private final InterruptableConsumer<T> consumer;
    private final BlockingQueue<T> queue  = new LinkedBlockingDeque<>(maxQueueSize);
    private final Thread thread = new Thread(this::process);

    {
        this.thread.start();
    }

    @FunctionalInterface
    public interface InterruptableConsumer<T> {
        void accept(List<T> t) throws InterruptedException;
    }
    
    private void process() {
        while (!Thread.interrupted()) {
            try {
                T item = queue.take();

                List<T> bucket = new ArrayList<>();
                bucket.add(item);

                while (bucket.size() < maxBatchSize && (item = queue.poll()) != null) {
                    bucket.add(item);
                }
                
                consumer.accept(bucket);
            } catch (InterruptedException e) {
                LOGGER.info("Interrupted");
                break;
            }
            catch (Exception ex) {
                LOGGER.info("Error while processing bucket queue", ex);
            }
        }
    }

    public boolean feed(T item) {
        return queue.offer(item);
    }
}