AndrewJSchofield commented on code in PR #15215:
URL: https://github.com/apache/kafka/pull/15215#discussion_r1459040148
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1134,9 +1134,22 @@ private CompletableFuture<Void> assignPartitions(
// Make assignment effective on the client by updating the
subscription state.
updateSubscription(assignedPartitions, false);
+ // Pause partitions to ensure that fetch does not start until the
callback completes.
+ assignedPartitions.forEach(tp ->
subscriptions.pause(tp.topicPartition()));
Review Comment:
I agree with this comment. There is a `KafkaConsumer.paused` method which
would show that we had paused the partitions internally, and that seems
incorrect in this case. They are logically not paused, but practically we are
not fetching records until the callback has been completed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
