ShivsundarR commented on code in PR #17061:
URL: https://github.com/apache/kafka/pull/17061#discussion_r1740642210
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -613,74 +607,99 @@ private void handleShareAcknowledgeSuccess(Node
fetchTarget,
ShareAcknowledgeRequestData
requestData,
AcknowledgeRequestState
acknowledgeRequestState,
ClientResponse resp,
- long currentTimeMs) {
+ long responseCompletionTimeMs) {
try {
log.debug("Completed ShareAcknowledge request from node {}
successfully", fetchTarget.id());
- final ShareAcknowledgeResponse response =
(ShareAcknowledgeResponse) resp.responseBody();
- final ShareSessionHandler handler =
acknowledgeRequestState.sessionHandler();
+ ShareAcknowledgeResponse response = (ShareAcknowledgeResponse)
resp.responseBody();
+
+ if (acknowledgeRequestState.onClose()) {
+ response.data().responses().forEach(topic ->
topic.partitions().forEach(partition -> {
+ TopicIdPartition tip = new
TopicIdPartition(topic.topicId(),
+ partition.partitionIndex(),
+ metadata.topicNames().get(topic.topicId()));
+ if (partition.errorCode() != Errors.NONE.code()) {
+
metricsManager.recordFailedAcknowledgements(acknowledgeRequestState.getInFlightAcknowledgementsCount(tip));
+ }
+ acknowledgeRequestState.handleAcknowledgeErrorCode(tip,
Errors.forCode(partition.errorCode()));
+ }));
- final short requestVersion = resp.requestHeader().apiVersion();
+
acknowledgeRequestState.onSuccessfulAttempt(responseCompletionTimeMs);
+ acknowledgeRequestState.processingComplete();
- if (!handler.handleResponse(response, requestVersion)) {
- acknowledgeRequestState.onFailedAttempt(currentTimeMs);
- if (response.error().exception() instanceof RetriableException
&& !acknowledgeRequestState.onClose()) {
- // We retry the request until the timer expires, unless we
are closing.
- acknowledgeRequestState.retryRequest();
+ if (!closeFuture.isDone()) {
+ closeFuture.complete(null);
+ }
+
+ metricsManager.recordLatency(resp.requestLatencyMs());
+ } else {
+ if
(!acknowledgeRequestState.sessionHandler.handleResponse(response,
resp.requestHeader().apiVersion())) {
+ // Received a response-level error code.
+
acknowledgeRequestState.onFailedAttempt(responseCompletionTimeMs);
+
+ if (response.error().exception() instanceof
RetriableException) {
+ // We retry the request until the timer expires,
unless we are closing.
+ acknowledgeRequestState.moveAllToIncompleteAcks();
+ } else {
+
response.data().responses().forEach(shareAcknowledgeTopicResponse ->
shareAcknowledgeTopicResponse.partitions().forEach(partitionData -> {
Review Comment:
Yes, we should clear the `inFlightAcknowledgements` for this case as well. I
have updated the PR, thanks!!
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
