dajac commented on code in PR #14557:
URL: https://github.com/apache/kafka/pull/14557#discussion_r1426733887
##########
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:
I wonder if we could have something like:
```
if (notRetriable(throwable) || !maybeRetry(currentTimeMs, throwable)) {
future.completeExceptionally(throwable);
}
```
Thinking about this because `retry` may not retry if the timer is expired.
--
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
