Ok, so the problem was that the code was trying to zip a map (topicsAndPartitions) and a list (messages). Even though they are derived from the same map, there is no guarantee that the map is going to be iterated in the same order as the list. So, in the future, we need to be careful about using zip. We need to make sure that the two collections to be zipped have well defined ordering (e.g., list) for iteration and the ordering is what we want. Could you double check other usage of zip introduced in the compression patch?
The patch looks fine. However, could you also add a unit test (probably just at the serialize() level) that exposes this problem?