#152: Added synchronization to the Stream.collect() consumer

...to synchronize access to the batch consumer internal state. A unit test illustrating the issue and the fix is added too.