Queue that connects multiple producers and multiple consumers The scenario is about processing 'Message' objects. Producer creates them and Consumer does the consumption. 

class Message{...}

class MessageBatch{

    public MessageBatch(Collection&lt;Message&gt; messages){...}

    public List&lt;Message&gt; getMessages(){...};

    //more utility methods like getMessageIds(), etc.
}


The following is a pipeline object (queue) that connects multiple producer threads and multiple consumer threads. Both my producers &amp; consumers are I/O bound (database) and that justifies this attempt.

API requirements


[A] Ability to add &amp; get batch of Message objects.
[B] Consumer has a way to realize that no more elements are expected.




public class MessagePipeline {

    //[1]; to support API requirement [B]
    public static final MessageBatch SENTINEL_BATCH = new MessageBatch(Collections.&lt;Message&gt;emptyList());

    private BlockingQueue&lt;Message&gt; messageQueue = new LinkedBlockingQueue&lt;&gt;();

    // to support API requirement [B]
    private CountDownLatch producersWorkCompletionLatch;

    public MessagePipeline(int numOfProducerThreads) {
        this.producersWorkCompletionLatch = new CountDownLatch(numOfProducerThreads);
    }

    public boolean add(MessageBatch batch) {
        isWriteAllowed(); // UPDATE: this step is misleading and thinking of removing this.
        return messageQueue.addAll(batch.getMessages());
    }

    // returns a batch containing up to maxSize elements. 
    public MessageBatch takeBatch(int maxSize) {
        List&lt;Message&gt; messages = drainIfMoreElementsExpected(new ArrayList&lt;Message&gt;(), maxSize);

        return messages.isEmpty() ? SENTINEL_BATCH : new MessageBatch(messages);
    }

    //[2]: Producer should invoke this at the end.
    public void notifyCompletion() {
        producersWorkCompletionLatch.countDown();
    }

    public boolean isMoreElementsExpected() {
        boolean isExpected = !(producersWorkCompletionLatch.getCount() == 0 &amp;&amp; messageQueue.size() == 0);
        return isExpected;
    }
    //[3] : This is CPU bound waiting. TODO: Make it I/O bound.
    private List&lt;Message&gt; drainIfMoreElementsExpected(List&lt;Message&gt; messages, int maxSize) {
        while (messages.isEmpty() &amp;&amp; isMoreElementsExpected()) {
            messageQueue.drainTo(messages, maxSize);
        }
        return messages;
    }

    /**
     * Assures write-operations 'happen-before' producer-threads-completion.
     */
    private void isWriteAllowed(){
        if(producersWorkCompletionLatch.getCount() == 0){
            throw new UnsupportedOperationException("New elements are not accepted after all producer threads notify of completion");
        }
    }
}


Questions  


I'm looking forward to code review in general.
Do approaches [1] and [2] seem right?
[3]: Am I right that it is CPU bound? How do I make it I/O bound waiting? Introducing locks seem to make it more complicated. 


I'm in the middle of writing unit tests. It's good to know if I missed anything because unit-testing multi-threaded code tells you that the code is broken only one in million times and doesn't tell you why!
Since you say that your whole system is based on message passing you should take a serious look at akka.   You get a rock solid message passing system, and you can even spread it on multiple machines in the future.

I have not checked in detail, but you have some very serious concurrency bugs.  For example, add checks if writing is allowed, then writes to the queue.  The problem is that you did not synchronize add, so the check might pass, but before calling the next line, some other thread might bring the count down latch down to zero.  BlockingQueue is thread-safe, but it is far from enough to make your code thread-safe.

I don't like the MessagePipeline api: it is too complex.  I really don't see the advantage of writing or taking "batches" of messages.  If the reader or writers want to process many messages, just let them read or write many messages themselves by processing their many messages in a loop.  The api itself should only read/write one message at a time.  In that case, your whole MessagePipeline is superfluous since a simple BlockingQueue is exactly what you want.  You will also avoid many concurrency bugs.  You would also need to keep the CountDownLatch or something similar to let the consumer know when it's over.