ShivsundarR commented on code in PR #18939:
URL: https://github.com/apache/kafka/pull/18939#discussion_r1961579512
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -851,59 +873,25 @@ private void handleShareAcknowledgeSuccess(Node
fetchTarget,
// We retry the request until the timer expires,
unless we are closing.
acknowledgeRequestState.moveAllToIncompleteAcks();
} else {
-
response.data().responses().forEach(shareAcknowledgeTopicResponse ->
shareAcknowledgeTopicResponse.partitions().forEach(partitionData -> {
- TopicIdPartition tip = new
TopicIdPartition(shareAcknowledgeTopicResponse.topicId(),
- partitionData.partitionIndex(),
-
metadata.topicNames().get(shareAcknowledgeTopicResponse.topicId()));
-
-
acknowledgeRequestState.handleAcknowledgeErrorCode(tip, response.error());
- }));
+
acknowledgeRequestState.processPendingInFlightAcknowledgements(response.error().exception());
acknowledgeRequestState.processingComplete();
}
+
} else {
AtomicBoolean shouldRetry = new AtomicBoolean(false);
// Check all partition level error codes
response.data().responses().forEach(shareAcknowledgeTopicResponse ->
shareAcknowledgeTopicResponse.partitions().forEach(partitionData -> {
Errors partitionError =
Errors.forCode(partitionData.errorCode());
- TopicIdPartition tip = new
TopicIdPartition(shareAcknowledgeTopicResponse.topicId(),
- partitionData.partitionIndex(),
-
metadata.topicNames().get(shareAcknowledgeTopicResponse.topicId()));
- if (partitionError.exception() != null) {
- boolean retry = false;
-
- if (partitionError ==
Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH) {
- // If the leader has changed, there's no point
in retrying the operation because the acquisition locks
- // will have been released.
- TopicPartition tp = new
TopicPartition(metadata.topicNames().get(shareAcknowledgeTopicResponse.topicId()),
partitionData.partitionIndex());
-
- log.debug("For {}, received error {}, with
leaderIdAndEpoch {}", tp, partitionError, partitionData.currentLeader());
- if (partitionData.currentLeader().leaderId()
!= -1 && partitionData.currentLeader().leaderEpoch() != -1) {
- partitionsWithUpdatedLeaderInfo.put(tp,
new Metadata.LeaderIdAndEpoch(
-
Optional.of(partitionData.currentLeader().leaderId()),
Optional.of(partitionData.currentLeader().leaderEpoch())));
- }
- } else if (partitionError.exception() instanceof
RetriableException) {
- retry = true;
- }
-
- if (retry) {
- // Move to incomplete acknowledgements to retry
-
acknowledgeRequestState.moveToIncompleteAcks(tip);
- shouldRetry.set(true);
- } else {
-
metricsManager.recordFailedAcknowledgements(acknowledgeRequestState.getInFlightAcknowledgementsCount(tip));
-
acknowledgeRequestState.handleAcknowledgeErrorCode(tip, partitionError);
- }
- } else {
-
acknowledgeRequestState.handleAcknowledgeErrorCode(tip, partitionError);
+ TopicIdPartition tip =
createTopicIdPartition(shareAcknowledgeTopicResponse.topicId(),
partitionData.partitionIndex());
+ if (tip == null) {
+ log.error("Invalid topic name in SA-2 response");
Review Comment:
Yeah, was using it for testing, thanks, I have updated it 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]
