KAFKA-14960: TopicMetadata request manager #14386
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -16,7 +16,6 @@ | |
| */ | ||
| package org.apache.kafka.clients.consumer.internals; | ||
|
|
||
| import org.apache.kafka.clients.consumer.ConsumerConfig; | ||
| import org.apache.kafka.clients.consumer.OffsetAndMetadata; | ||
| import org.apache.kafka.clients.consumer.RetriableCommitFailedException; | ||
|
|
@@ -27,7 +26,6 @@ | |
| import org.apache.kafka.common.message.OffsetCommitRequestData; | ||
| import org.apache.kafka.common.protocol.Errors; | ||
| import org.apache.kafka.common.record.RecordBatch; | ||
| import org.apache.kafka.common.requests.OffsetFetchRequest; | ||
| import org.apache.kafka.common.requests.OffsetFetchResponse; | ||
| import org.apache.kafka.common.utils.LogContext; | ||
|
|
@@ -91,7 +89,7 @@ public CommitRequestManager( | |
| } | ||
|
|
||
| /** | ||
| * Poll for the {@link OffsetFetchRequest} and {@link org.apache.kafka.common.requests.OffsetCommitRequest} request if there's any. The function will | ||
| * also try to autocommit the offsets, if feature is enabled. | ||
| */ | ||
| @Override | ||
|
|
@@ -106,6 +104,7 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { | |
| return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList()); | ||
| } | ||
|
|
||
| pendingRequests.inflightOffsetFetches.forEach(System.out::println); | ||
|
|
||
| return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, | ||
| Collections.unmodifiableList(pendingRequests.drain(currentTimeMs))); | ||
| } | ||
|
|
@@ -127,9 +126,9 @@ public void maybeAutoCommit(final Map<TopicPartition, OffsetAndMetadata> offsets | |
|
|
||
| /** | ||
| * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an | ||
| * {@link OffsetCommitRequest} and enqueue it to send later. | ||
| */ | ||
| public CompletableFuture<Void> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) { | ||
| return pendingRequests.addOffsetCommitRequest(offsets); | ||
| } | ||
|
|
||
|
|
@@ -145,19 +144,18 @@ public void updateAutoCommitTimer(final long currentTimeMs) { | |
| this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs)); | ||
| } | ||
|
|
||
| // Visible for testing | ||
| List<OffsetFetchRequestState> unsentOffsetFetchRequests() { | ||
| return pendingRequests.unsentOffsetFetches; | ||
| } | ||
|
|
||
| // Visible for testing | ||
| Queue<OffsetCommitRequest> unsentOffsetCommitRequests() { | ||
| return pendingRequests.unsentOffsetCommits; | ||
| } | ||
|
|
||
| // Visible for testing | ||
| CompletableFuture<Void> sendAutoCommit(final Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets) { | ||
| log.debug("Enqueuing autocommit offsets: {}", allConsumedOffsets); | ||
| return this.addOffsetCommitRequest(allConsumedOffsets) | ||
| .whenComplete((response, throwable) -> { | ||
|
|
@@ -177,28 +175,24 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse | |
| }); | ||
| } | ||
|
|
||
| private class OffsetCommitRequest { | ||
| private final Map<TopicPartition, OffsetAndMetadata> offsets; | ||
| private final String groupId; | ||
| private final GroupState.Generation generation; | ||
| private final String groupInstanceId; | ||
| private final CompletableFuture<Void> future; | ||
|
|
||
| public OffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets, | ||
| final String groupId, | ||
| final String groupInstanceId, | ||
| final GroupState.Generation generation) { | ||
| this.offsets = offsets; | ||
| this.future = new CompletableFuture<>(); | ||
| this.groupId = groupId; | ||
| this.generation = generation; | ||
| this.groupInstanceId = groupInstanceId; | ||
| } | ||
|
|
||
| public NetworkClientDelegate.UnsentRequest toUnsentRequest() { | ||
| Map<String, OffsetCommitRequestData.OffsetCommitRequestTopic> requestTopicDataMap = new HashMap<>(); | ||
| for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) { | ||
|
|
@@ -220,7 +214,7 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() { | |
| requestTopicDataMap.put(topicPartition.topic(), topic); | ||
| } | ||
|
|
||
| org.apache.kafka.common.requests.OffsetCommitRequest.Builder builder = new org.apache.kafka.common.requests.OffsetCommitRequest.Builder( | ||
|
There was a problem hiding this comment. Could we import the package? |
||
| new OffsetCommitRequestData() | ||
| .setGroupId(this.groupId) | ||
| .setGenerationIdOrMemberEpoch(generation.generationId) | ||
|
|
@@ -230,15 +224,20 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() { | |
| return new NetworkClientDelegate.UnsentRequest( | ||
| builder, | ||
| coordinatorRequestManager.coordinator(), | ||
| (response, throwable) -> { | ||
| if (throwable == null) { | ||
| this.future.complete(null); | ||
| } else { | ||
| this.future.completeExceptionally(throwable); | ||
| } | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| private class OffsetFetchRequestState extends RequestState { | ||
| public final Set<TopicPartition> requestedPartitions; | ||
| public final GroupState.Generation requestedGeneration; | ||
| CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future; | ||
|
|
||
| public OffsetFetchRequestState(final Set<TopicPartition> partitions, | ||
| final GroupState.Generation generation, | ||
| final long retryBackoffMs, | ||
|
|
@@ -259,13 +258,10 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTim | |
| true, | ||
| new ArrayList<>(this.requestedPartitions), | ||
| throwOnFetchStableOffsetUnsupported); | ||
| return new NetworkClientDelegate.UnsentRequest( | ||
| builder, | ||
| coordinatorRequestManager.coordinator(), | ||
| (r, t) -> onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody())); | ||
| } | ||
|
|
||
| public void onResponse( | ||
|
|
@@ -298,8 +294,8 @@ private void onFailure(final long currentTimeMs, | |
| } | ||
|
|
||
| private void retry(final long currentTimeMs) { | ||
| this.onFailedAttempt(currentTimeMs); | ||
|
There was a problem hiding this comment. currentTimeMs is when the request is sent, not when the response is receive. Could we pass along There was a problem hiding this comment. Thanks for catching this. There was a problem hiding this comment. Addressed, see line #259 |
||
| this.onSendAttempt(currentTimeMs); | ||
| pendingRequests.addOffsetFetchRequest(this); | ||
| } | ||
|
|
||
|
|
@@ -359,12 +355,12 @@ private void onSuccess(final long currentTimeMs, | |
| } | ||
| } | ||
|
|
||
| private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> otherFuture) { | ||
| return this.future.whenComplete((r, t) -> { | ||
| if (t != null) { | ||
| otherFuture.completeExceptionally(t); | ||
| } else { | ||
| otherFuture.complete(r); | ||
| } | ||
| }); | ||
| } | ||
|
|
@@ -381,41 +377,41 @@ public String toString() { | |
| } | ||
|
|
||
| /** | ||
| * <p>This is used to stage the unsent {@link OffsetCommitRequest} and {@link OffsetFetchRequestState}. | ||
| * <li>unsentOffsetCommits holds the offset commit requests that have not been sent out</> | ||
| * <li>unsentOffsetFetches holds the offset fetch requests that have not been sent out</li> | ||
| * <li>inflightOffsetFetches holds the offset fetch requests that have been sent out but not completed</>. | ||
| * <p> | ||
| * {@code addOffsetFetchRequest} dedupes the requests to avoid sending the same requests. | ||
| */ | ||
|
|
||
| class PendingRequests { | ||
| // Queue is used to ensure the sequence of commit | ||
| Queue<OffsetCommitRequest> unsentOffsetCommits = new LinkedList<>(); | ||
| List<OffsetFetchRequestState> unsentOffsetFetches = new ArrayList<>(); | ||
| List<OffsetFetchRequestState> inflightOffsetFetches = new ArrayList<>(); | ||
|
|
||
| boolean hasUnsentRequests() { | ||
|
There was a problem hiding this comment. Could this be private? Ditto below. There was a problem hiding this comment. Sorry - I left out commit. Fixing those obvious mistakes right away. There was a problem hiding this comment. Sorry - leaving this protected as it is used in one of the test to verify there's no more unsent request. |
||
| return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty(); | ||
| } | ||
|
|
||
| CompletableFuture<Void> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) { | ||
| // TODO: Dedupe committing the same offsets to the same partitions | ||
| OffsetCommitRequest request = new OffsetCommitRequest( | ||
| offsets, | ||
| groupState.groupId, | ||
| groupState.groupInstanceId.orElse(null), | ||
| groupState.generation); | ||
| unsentOffsetCommits.add(request); | ||
| return request.future; | ||
| } | ||
|
|
||
| /** | ||
| * <p>Adding an offset fetch request to the outgoing buffer. If the same request was made, we chain the future | ||
| * to the existing one. | ||
| * | ||
| * <p>If the request is new, it invokes a callback to remove itself from the {@code inflightOffsetFetches} | ||
| * upon completion.</> | ||
|
There was a problem hiding this comment. nit: wrong tag </> and other unclosed ones above |
||
| */ | ||
| private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final OffsetFetchRequestState request) { | ||
| Optional<OffsetFetchRequestState> dupe = | ||
|
|
@@ -451,17 +447,17 @@ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetch | |
| /** | ||
| * Clear {@code unsentOffsetCommits} and moves all the sendable request in {@code unsentOffsetFetches} to the | ||
| * {@code inflightOffsetFetches} to bookkeep all of the inflight requests. | ||
| * <p> | ||
|
There was a problem hiding this comment. was the intention here to add an extra line in-between? should be There was a problem hiding this comment. 😓 - must be the residual from some edits.. |
||
| * Note: Sendable requests are determined by their timer as we are expecting backoff on failed attempt. See | ||
| * {@link RequestState}. | ||
| **/ | ||
| List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) { | ||
| List<NetworkClientDelegate.UnsentRequest> unsentRequests = new ArrayList<>(); | ||
|
|
||
| // Add all unsent offset commit requests to the unsentRequests list | ||
| unsentRequests.addAll( | ||
| unsentOffsetCommits.stream() | ||
| .map(OffsetCommitRequest::toUnsentRequest) | ||
| .collect(Collectors.toList())); | ||
|
|
||
| // Partition the unsent offset fetch requests into sendable and non-sendable lists | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we need to spell out the package name for OffsetCommitRequest? It doesn't seem this is done consistently.