AndrewJSchofield commented on code in PR #23499:
URL: https://github.com/apache/kafka/pull/23499#discussion_r4061048131
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:
##########
@@ -227,11 +237,16 @@ 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:
I am not going to add this into this PR. I am going to make a couple of
larger changes in this area as a result of this issue, but not in this PR. I
plan to eradicate the side-effect of cleaning up batches from the getter
`numRecords` but that will require replacing the pruning code with a different
trigger, and I will sort out the situation in which we have batches for the
same topic-partition for previous and new partition leader.
--
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]
