private static void processSink(final Socket connectionSocket) {
        //increment the sink count
        sinkCount.addAndGet(1);
        //create a new sink task
        Runnable sinkTask = new Runnable() {
            @Override
            public void run() {
                //get sink socket
                Socket sinkSocket = connectionSocket;
                while (listening) {
                    //output stream to sink
                    DataOutputStream outToSink = null;
                    try {
                        outToSink = new DataOutputStream(
                                sinkSocket.getOutputStream());

                        if (!messages.isEmpty()) {
                            //send source messages to sink
                            for (int i = 0; i < messages.size(); i++) {
                                outToSink.writeBytes(messages.get(i));
                               
                            }
                             messages.clear();
                        } else {
                            synchronized (lock) {
                                //we wait for messages to be produced by the sources
                                lock.wait();
                            }
                        }
                    } catch (IOException e) {
                        break;
                    } catch (InterruptedException e) {
                        break;
                    }
                }
            }
        };
        //run the source task via executor
        exec.execute(sinkTask);
    }