ShivsundarR commented on code in PR #16727:
URL: https://github.com/apache/kafka/pull/16727#discussion_r1698620806
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -378,32 +452,38 @@ public CompletableFuture<Void> acknowledgeOnClose(final
Map<TopicIdPartition, Ac
acknowledgementsMapForNode.put(tip, acknowledgements);
metricsManager.recordAcknowledgementSent(acknowledgements.size());
- log.debug("Added closing acknowledge request for
partition {} to node {}", tip.topicPartition(), node.id());
+ log.debug("Added closing acknowledge request for
partition {} to node {}", tip.topicPartition(),
+ node.id());
resultCount.incrementAndGet();
}
}
- acknowledgeRequestStates.add(new
AcknowledgeRequestState(logContext,
- ShareConsumeRequestManager.class.getSimpleName() +
":3",
- deadlineMs,
- retryBackoffMs,
- retryBackoffMaxMs,
- sessionHandler,
- nodeId,
- acknowledgementsMapForNode,
- this::handleShareAcknowledgeCloseSuccess,
- this::handleShareAcknowledgeCloseFailure,
- resultHandler,
- true
- ));
+
+ acknowledgeRequestStates.putIfAbsent(nodeId, new Pair<>(null,
null));
+ // There can only be one commitSync()/close() happening at a
time. So per node,
Review Comment:
Right yeah have added a check now.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -287,22 +338,29 @@ public CompletableFuture<Map<TopicIdPartition,
Acknowledgements>> commitSync(
acknowledgementsMapForNode.put(tip, acknowledgements);
metricsManager.recordAcknowledgementSent(acknowledgements.size());
- log.debug("Added sync acknowledge request for
partition {} to node {}", tip.topicPartition(), node.id());
+ log.debug("Added sync acknowledge request for
partition {} to node {}", tip.topicPartition(),
+ node.id());
resultCount.incrementAndGet();
}
}
- acknowledgeRequestStates.add(new
AcknowledgeRequestState(logContext,
- ShareConsumeRequestManager.class.getSimpleName() +
":1",
- deadlineMs,
- retryBackoffMs,
- retryBackoffMaxMs,
- sessionHandler,
- nodeId,
- acknowledgementsMapForNode,
- this::handleShareAcknowledgeSuccess,
- this::handleShareAcknowledgeFailure,
- resultHandler
- ));
+ acknowledgeRequestStates.putIfAbsent(nodeId, new Pair<>(null,
null));
+
+ // There can only be one commitSync()/close() happening at a
time. So per node,
Review Comment:
Yes makes sense, have added a check now.
--
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]
