AndrewJSchofield commented on code in PR #23499:
URL: https://github.com/apache/kafka/pull/23499#discussion_r4045415534
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -227,11 +236,17 @@ public boolean checkAllInFlightAreAcknowledged() {
*/
public Map<TopicIdPartition, NodeAcknowledgements>
takeAcknowledgedRecords() {
Map<TopicIdPartition, NodeAcknowledgements> acknowledgementMap = new
LinkedHashMap<>();
- batches.forEach((tip, batch) -> {
- int nodeId = batch.nodeId();
- Acknowledgements acknowledgements =
batch.takeAcknowledgedRecords();
- if (!acknowledgements.isEmpty())
- acknowledgementMap.put(tip, new NodeAcknowledgements(nodeId,
acknowledgements));
+ batches.forEach((tip, batchList) -> {
+ if (!batchList.isEmpty()) {
+ Acknowledgements acknowledgements = Acknowledgements.empty();
+ int nodeId = batchList.get(0).nodeId();
Review Comment:
That is true in principle, and the same was true for the code which merged
the batches. If the leader has changed, the acks will all fail on the old
leader, and the acquired records on the new leader will time out, and then it
will continue as expected. I do plan to resolve this too, but I don't want to
grow this PR too much because it is intended for a release candidate. To be
clear, this is only going to happen when there are multiple batches buffered up
for a topic-partition and a leader change too, and that's not a common scenario.
--
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]
