Looking again at the collector code, I believe it's still not correct:

1. Thread T1 enters combineAndMigrate, acquires the collector lock and starts iterating over the collected values map. It finds entry k1 = [v1]
2. Thread T2 enters collector.emit(k1, v2). It also finds the list k1 = [v1] in the store.
3. T1 proceeds removes entry k = [v1] from the collector.
4. T2 now adds v2 to the k = [v1] entry.

I'm not going to hazard myself to suggest the "correct" solution again, but I think collecting into a simple queue and having a single thread that splits the intermediate values based on the key might be better. In any case, I suggest extracting the collector in a top-level class and adding a stress test for it.