Synchronizing on every key is bad, and I don't think it's really needed. We can allow the emitCount and the actual map size to get out of sync a bit, so we could have a method like this in DefaultCollector:

      public Map<KOut, List<VOut>> removeChunk(int chunkSize) {
         if (emitCount.get() <= chunkSize)
            return null;

         emitCount.set(0);
         ConcurrentMap<KOut, List<VOut>> oldStore = store;
         store = CollectionFactory.makeConcurrentMap();
         return oldStore;
      }

We could also remove emitCount completely, it's true that `EquivalentConsistentHashV8.size()` it's a little more costly but it compensates by avoiding a the write contention when incrementing the size.

BTW, I didn't see any code resetting emitCount. You may need some more tests :)