Instead of maintaining a hashmap (like for the current bundlers), a simple blocking bounded queue of messages would be used. Whenever we've reached M bytes or N milliseconds have elapsed, a consumer thread processes the queue in the following manner:

• First set the queue to a new queue (volatile assignment), reuse a number of queues
• Iterate through all messages in the current queue, for each message:
  • If the destination is the same as the current destination, write the message to the stream for the current destination
  • Else set the current destination to msg.getDest() and create an output stream (similar to writing a message list)
    • Stream the current message to the output stream
    • If there was a previous destination, close the associated output stream and send the message list

Example:

• We have messages with the following destinations: A, null, null, B, B, null, A, null, null, null
• First we send a message list consisting of 1 message to A
• Next we send a message list consisting of 2 messages to the cluster (dest==null)
• Then we send a batch of 2 messages to B, 1 to the cluster, 1 to A and 3 to the cluster