chia7712 commented on code in PR #17061:
URL: https://github.com/apache/kafka/pull/17061#discussion_r1740569415
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -982,6 +941,10 @@ boolean maybeExpire() {
return numAttempts > 0 && isExpired();
}
+ /**
+ * Moves the in-flight acknowledgements for a given partition to
incomplete acknowledgements to retry
+ * in the next request.
+ */
public void moveToIncompleteAcks(TopicIdPartition tip) {
Acknowledgements acks = inFlightAcknowledgements.remove(tip);
if (acks != null) {
Review Comment:
Why it needs to check the existent acks from `incompleteAcknowledgements`?
It seems `incompleteAcknowledgements` will be empty in this case. By contrast,
`moveAllToIncompleteAcks` have no such behavior of checking existent acks.
##########
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:
Should we clear the `inFlightAcknowledgements` when the response-level error
is not retryable?
--
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
