lianetm commented on code in PR #14557:
URL: https://github.com/apache/kafka/pull/14557#discussion_r1427025116
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -381,23 +441,37 @@ public NetworkClientDelegate.UnsentRequest
toUnsentRequest() {
requestTopicDataMap.put(topicPartition.topic(), topic);
}
- OffsetCommitRequest.Builder builder = new
OffsetCommitRequest.Builder(
- new OffsetCommitRequestData()
+ OffsetCommitRequestData data = new OffsetCommitRequestData()
.setGroupId(this.groupId)
- .setGenerationIdOrMemberEpoch(generation.generationId)
- .setMemberId(generation.memberId)
- .setGroupInstanceId(groupInstanceId)
- .setTopics(new ArrayList<>(requestTopicDataMap.values())));
+ .setGroupInstanceId(groupInstanceId.orElse(null))
+ .setTopics(new ArrayList<>(requestTopicDataMap.values()));
+ if (memberInfo.memberId.isPresent()) {
+ data = data.setMemberId(memberInfo.memberId.get());
+ }
+ if (memberInfo.memberEpoch.isPresent()) {
+ data =
data.setGenerationIdOrMemberEpoch(memberInfo.memberEpoch.get());
+ }
+
+ OffsetCommitRequest.Builder builder = new
OffsetCommitRequest.Builder(data);
+
NetworkClientDelegate.UnsentRequest resp = new
NetworkClientDelegate.UnsentRequest(
builder,
coordinatorRequestManager.coordinator());
resp.whenComplete(
(response, throwable) -> {
try {
if (throwable == null) {
+ log.debug("OffsetCommit response received for
offsets {} ", offsets);
onResponse(response);
} else {
- onError(throwable,
resp.handler().completionTimeMs());
+ log.debug("OffsetCommit completed with error for
offsets {}", offsets, throwable);
+ long currentTimeMs =
resp.handler().completionTimeMs();
+ handleCoordinatorDisconnect(throwable,
currentTimeMs);
+ if (throwable instanceof RetriableException) {
+ retry(currentTimeMs, throwable);
+ } else {
+ future.completeExceptionally(throwable);
+ }
Review Comment:
The `retry` already fails the request future if it does not retry (ex. timer
expired). That's needed from all the places where we call the retry, so seemed
sensible to keep it centralized there.
--
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
