Thanks for the new patch. Some more comments -
1.1 Remove unused import import java.util.concurrent.atomic.AtomicInteger
1.2 I thought you wanted to clear the map in the build method. Don't see that change included in this patch ?
2.1 It is not ideal to have the SimpleConsumer set the correlation id correctly on each fetch request. It is better to leave it as it was done before (increment the correlation id in build()) and just change it to use Utils.getNextNonNegativeInt.