You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "philipnee (via GitHub)" <gi...@apache.org> on 2023/03/11 22:19:16 UTC
[GitHub] [kafka] philipnee opened a new pull request, #13380: Consumer refactor committed api
philipnee opened a new pull request, #13380:
URL: https://github.com/apache/kafka/pull/13380
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1143865476
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -113,15 +112,45 @@ private void maybeAutoCommit(final long currentTimeMs) {
}
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptionState.allConsumed();
- log.debug("Auto-committing offsets {}", allConsumedOffsets);
sendAutoCommit(allConsumedOffsets);
autocommit.resetTimer();
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
+ * {@link UnsentOffsetCommitRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ return pendingRequests.addOffsetCommitRequest(offsets);
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ return pendingRequests.addOffsetFetchRequest(partitions);
+ }
+
+ public void clientPoll(final long currentTimeMs) {
+ this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
+ }
+
+ // Visible for testing
+ List<UnsentOffsetFetchRequest> unsentOffsetFetchRequests() {
+ return pendingRequests.unsentOffsetFetches;
+ }
+
+ // Visible for testing
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommitRequests() {
+ return pendingRequests.unsentOffsetCommits;
}
// Visible for testing
CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets) {
- CompletableFuture<ClientResponse> future = this.add(allConsumedOffsets)
+ log.debug("Enqueuing autocommit offsets: {}", allConsumedOffsets);
+ return this.addOffsetCommitRequest(allConsumedOffsets)
Review Comment:
I think you are right, we could just use the pendingQueue to handle the inflight auto commit. Thanks for spotting this.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1150022881
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -61,73 +83,250 @@ public void setup() {
}
@Test
- public void testPoll() {
+ public void testPoll_EnsureManualCommitSent() {
CommitRequestManager commitRequestManger = create(false, 0);
- NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- assertEquals(0, res.unsentRequests.size());
- assertEquals(Long.MAX_VALUE, res.timeUntilNextPollMs);
+ assertPoll(0, commitRequestManger);
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0));
- commitRequestManger.add(offsets);
- res = commitRequestManger.poll(time.milliseconds());
- assertEquals(1, res.unsentRequests.size());
+ commitRequestManger.addOffsetCommitRequest(offsets);
+ assertPoll(1, commitRequestManger);
}
@Test
- public void testPollAndAutoCommit() {
+ public void testPoll_EnsureAutocommitSent() {
CommitRequestManager commitRequestManger = create(true, 100);
- NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- assertEquals(0, res.unsentRequests.size());
- assertEquals(Long.MAX_VALUE, res.timeUntilNextPollMs);
+ assertPoll(0, commitRequestManger);
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0));
- commitRequestManger.clientPoll(time.milliseconds());
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
when(subscriptionState.allConsumed()).thenReturn(offsets);
time.sleep(100);
- commitRequestManger.clientPoll(time.milliseconds());
- res = commitRequestManger.poll(time.milliseconds());
- assertEquals(1, res.unsentRequests.size());
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ assertPoll(1, commitRequestManger);
}
@Test
- public void testAutocommitStateUponFailure() {
+ public void testPoll_EnsureCorrectInflightRequestBufferSize() {
+ CommitRequestManager commitManager = create(false, 100);
+
+ // Create some offset commit requests
+ Map<TopicPartition, OffsetAndMetadata> offsets1 = new HashMap<>();
+ offsets1.put(new TopicPartition("test", 0), new OffsetAndMetadata(10L));
+ offsets1.put(new TopicPartition("test", 1), new OffsetAndMetadata(20L));
+ Map<TopicPartition, OffsetAndMetadata> offsets2 = new HashMap<>();
+ offsets2.put(new TopicPartition("test", 3), new OffsetAndMetadata(20L));
+ offsets2.put(new TopicPartition("test", 4), new OffsetAndMetadata(20L));
+
+ // Add the requests to the CommitRequestManager and store their futures
+ ArrayList<CompletableFuture<ClientResponse>> commitFutures = new ArrayList<>();
+ ArrayList<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> fetchFutures = new ArrayList<>();
+ commitFutures.add(commitManager.addOffsetCommitRequest(offsets1));
+ fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 0))));
+ commitFutures.add(commitManager.addOffsetCommitRequest(offsets2));
+ fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 1))));
+
+ // Poll the CommitRequestManager and verify that the inflightOffsetFetches size is correct
+ NetworkClientDelegate.PollResult result = commitManager.poll(time.milliseconds());
+ assertEquals(4, result.unsentRequests.size());
+ assertTrue(result.unsentRequests
+ .stream().anyMatch(r -> r.requestBuilder() instanceof OffsetCommitRequest.Builder));
+ assertTrue(result.unsentRequests
+ .stream().anyMatch(r -> r.requestBuilder() instanceof OffsetFetchRequest.Builder));
+ assertFalse(commitManager.pendingRequests.hasUnsentRequests());
+ assertEquals(2, commitManager.pendingRequests.inflightOffsetFetches.size());
+
+ // Verify that the inflight offset fetch requests have been removed from the pending request buffer
+ commitFutures.forEach(f -> f.complete(null));
+ fetchFutures.forEach(f -> f.complete(null));
+ assertEquals(0, commitManager.pendingRequests.inflightOffsetFetches.size());
+ }
+
+ @Test
+ public void testPoll_EnsureEmptyPendingRequestAfterPoll() {
CommitRequestManager commitRequestManger = create(true, 100);
- time.sleep(100);
- commitRequestManger.clientPoll(time.milliseconds());
- NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- time.sleep(100);
- // We want to make sure we don't resend autocommit if the previous request has not been completed
- assertEquals(Long.MAX_VALUE, commitRequestManger.poll(time.milliseconds()).timeUntilNextPollMs);
+ commitRequestManger.addOffsetCommitRequest(new HashMap<>());
+ assertEquals(1, commitRequestManger.unsentOffsetCommitRequests().size());
+ commitRequestManger.poll(time.milliseconds());
+ assertTrue(commitRequestManger.unsentOffsetCommitRequests().isEmpty());
+ assertEmptyPendingRequests(commitRequestManger);
+ }
+ @Test
+ public void testAutocommit_ResendAutocommitAfterException() {
+ CommitRequestManager commitRequestManger = create(true, 100);
+ time.sleep(100);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ List<CompletableFuture<ClientResponse>> futures = assertPoll(1, commitRequestManger);
+ time.sleep(99);
// complete the autocommit request (exceptionally)
- res.unsentRequests.get(0).future().completeExceptionally(new KafkaException("test exception"));
+ futures.get(0).completeExceptionally(new KafkaException("test exception"));
// we can then autocommit again
- commitRequestManger.clientPoll(time.milliseconds());
- res = commitRequestManger.poll(time.milliseconds());
- assertEquals(1, res.unsentRequests.size());
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ assertPoll(0, commitRequestManger);
+ time.sleep(1);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ assertPoll(1, commitRequestManger);
+ assertEmptyPendingRequests(commitRequestManger);
}
@Test
- public void testEnsureStagedCommitsPurgedAfterPoll() {
+ public void testAutocommit_EnsureOnlyOneInflightRequest() {
CommitRequestManager commitRequestManger = create(true, 100);
- commitRequestManger.add(new HashMap<>());
- assertEquals(1, commitRequestManger.stagedCommits().size());
- NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- assertTrue(commitRequestManger.stagedCommits().isEmpty());
+ time.sleep(100);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ List<CompletableFuture<ClientResponse>> futures = assertPoll(1, commitRequestManger);
+ time.sleep(100);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
Review Comment:
@guozhangwang - Fixed setting the inflight request in the `maybeAutoCommit` method. This test should address this issue/comment.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -61,73 +83,250 @@ public void setup() {
}
@Test
- public void testPoll() {
+ public void testPoll_EnsureManualCommitSent() {
CommitRequestManager commitRequestManger = create(false, 0);
- NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- assertEquals(0, res.unsentRequests.size());
- assertEquals(Long.MAX_VALUE, res.timeUntilNextPollMs);
+ assertPoll(0, commitRequestManger);
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0));
- commitRequestManger.add(offsets);
- res = commitRequestManger.poll(time.milliseconds());
- assertEquals(1, res.unsentRequests.size());
+ commitRequestManger.addOffsetCommitRequest(offsets);
+ assertPoll(1, commitRequestManger);
}
@Test
- public void testPollAndAutoCommit() {
+ public void testPoll_EnsureAutocommitSent() {
CommitRequestManager commitRequestManger = create(true, 100);
- NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- assertEquals(0, res.unsentRequests.size());
- assertEquals(Long.MAX_VALUE, res.timeUntilNextPollMs);
+ assertPoll(0, commitRequestManger);
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0));
- commitRequestManger.clientPoll(time.milliseconds());
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
when(subscriptionState.allConsumed()).thenReturn(offsets);
time.sleep(100);
- commitRequestManger.clientPoll(time.milliseconds());
- res = commitRequestManger.poll(time.milliseconds());
- assertEquals(1, res.unsentRequests.size());
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ assertPoll(1, commitRequestManger);
}
@Test
- public void testAutocommitStateUponFailure() {
+ public void testPoll_EnsureCorrectInflightRequestBufferSize() {
+ CommitRequestManager commitManager = create(false, 100);
+
+ // Create some offset commit requests
+ Map<TopicPartition, OffsetAndMetadata> offsets1 = new HashMap<>();
+ offsets1.put(new TopicPartition("test", 0), new OffsetAndMetadata(10L));
+ offsets1.put(new TopicPartition("test", 1), new OffsetAndMetadata(20L));
+ Map<TopicPartition, OffsetAndMetadata> offsets2 = new HashMap<>();
+ offsets2.put(new TopicPartition("test", 3), new OffsetAndMetadata(20L));
+ offsets2.put(new TopicPartition("test", 4), new OffsetAndMetadata(20L));
+
+ // Add the requests to the CommitRequestManager and store their futures
+ ArrayList<CompletableFuture<ClientResponse>> commitFutures = new ArrayList<>();
+ ArrayList<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> fetchFutures = new ArrayList<>();
+ commitFutures.add(commitManager.addOffsetCommitRequest(offsets1));
+ fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 0))));
+ commitFutures.add(commitManager.addOffsetCommitRequest(offsets2));
+ fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 1))));
+
+ // Poll the CommitRequestManager and verify that the inflightOffsetFetches size is correct
+ NetworkClientDelegate.PollResult result = commitManager.poll(time.milliseconds());
+ assertEquals(4, result.unsentRequests.size());
+ assertTrue(result.unsentRequests
+ .stream().anyMatch(r -> r.requestBuilder() instanceof OffsetCommitRequest.Builder));
+ assertTrue(result.unsentRequests
+ .stream().anyMatch(r -> r.requestBuilder() instanceof OffsetFetchRequest.Builder));
+ assertFalse(commitManager.pendingRequests.hasUnsentRequests());
+ assertEquals(2, commitManager.pendingRequests.inflightOffsetFetches.size());
+
+ // Verify that the inflight offset fetch requests have been removed from the pending request buffer
+ commitFutures.forEach(f -> f.complete(null));
+ fetchFutures.forEach(f -> f.complete(null));
+ assertEquals(0, commitManager.pendingRequests.inflightOffsetFetches.size());
+ }
+
+ @Test
+ public void testPoll_EnsureEmptyPendingRequestAfterPoll() {
CommitRequestManager commitRequestManger = create(true, 100);
- time.sleep(100);
- commitRequestManger.clientPoll(time.milliseconds());
- NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- time.sleep(100);
- // We want to make sure we don't resend autocommit if the previous request has not been completed
- assertEquals(Long.MAX_VALUE, commitRequestManger.poll(time.milliseconds()).timeUntilNextPollMs);
+ commitRequestManger.addOffsetCommitRequest(new HashMap<>());
+ assertEquals(1, commitRequestManger.unsentOffsetCommitRequests().size());
+ commitRequestManger.poll(time.milliseconds());
+ assertTrue(commitRequestManger.unsentOffsetCommitRequests().isEmpty());
+ assertEmptyPendingRequests(commitRequestManger);
+ }
+ @Test
+ public void testAutocommit_ResendAutocommitAfterException() {
+ CommitRequestManager commitRequestManger = create(true, 100);
+ time.sleep(100);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ List<CompletableFuture<ClientResponse>> futures = assertPoll(1, commitRequestManger);
+ time.sleep(99);
// complete the autocommit request (exceptionally)
- res.unsentRequests.get(0).future().completeExceptionally(new KafkaException("test exception"));
+ futures.get(0).completeExceptionally(new KafkaException("test exception"));
// we can then autocommit again
- commitRequestManger.clientPoll(time.milliseconds());
- res = commitRequestManger.poll(time.milliseconds());
- assertEquals(1, res.unsentRequests.size());
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ assertPoll(0, commitRequestManger);
+ time.sleep(1);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ assertPoll(1, commitRequestManger);
+ assertEmptyPendingRequests(commitRequestManger);
}
@Test
- public void testEnsureStagedCommitsPurgedAfterPoll() {
+ public void testAutocommit_EnsureOnlyOneInflightRequest() {
CommitRequestManager commitRequestManger = create(true, 100);
- commitRequestManger.add(new HashMap<>());
- assertEquals(1, commitRequestManger.stagedCommits().size());
- NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- assertTrue(commitRequestManger.stagedCommits().isEmpty());
+ time.sleep(100);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ List<CompletableFuture<ClientResponse>> futures = assertPoll(1, commitRequestManger);
+ time.sleep(100);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+ // We want to make sure we don't resend autocommit if the previous request has not been completed
+ assertPoll(0, commitRequestManger);
+ assertEmptyPendingRequests(commitRequestManger);
+
+ // complete the unsent request and re-poll
+ futures.get(0).complete(null);
+ assertPoll(1, commitRequestManger);
}
@Test
- public void testAutoCommitFuture() {
+ public void testOffsetFetchRequest_EnsureDuplicatedRequestSucceed() {
Review Comment:
I renamed the test names to elucidate the intents of different tests
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -87,10 +102,58 @@ public void testSuccessfulStartupShutdown() {
}
@Test
- public void testCommitAsync() {
+ public void testInvalidGroupId() {
+ this.groupId = null;
consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
- consumer.commitAsync();
- verify(eventHandler).add(any());
+ assertThrows(InvalidGroupIdException.class, () -> consumer.committed(new HashSet<>()));
+ }
+
+ @Test
+ public void testCommitAsync_NullCallback() throws InterruptedException {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(new TopicPartition("my-topic", 0), new OffsetAndMetadata(100L));
+ offsets.put(new TopicPartition("my-topic", 1), new OffsetAndMetadata(200L));
+
+ PrototypeAsyncConsumer<?, ?> mockedConsumer = spy(newConsumer(time, new StringDeserializer(), new StringDeserializer()));
+ doReturn(future).when(mockedConsumer).commit(offsets);
+ mockedConsumer.commitAsync(offsets, null);
+ future.complete(null);
+ TestUtils.waitForCondition(() -> future.isDone(),
+ 2000,
+ "commit future should complete");
+
+ assertFalse(future.isCompletedExceptionally());
+ }
+
+
+ @Test
+ public void testCommitAsync_UserSuppliedCallback() {
Review Comment:
@guozhangwang - This test and the above are to address your comment about null DefaultCommitCallback.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1133346239
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/OffsetFetchApplicationEvent.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.events;
+
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class OffsetFetchApplicationEvent extends ApplicationEvent {
+ final public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+ public final Set<TopicPartition> partitions;
+
+ public OffsetFetchApplicationEvent(final Set<TopicPartition> partitions) {
+ super(Type.FETCH_COMMITTED_OFFSET);
+ this.partitions = partitions;
+ this.future = new CompletableFuture<>();
+ }
+
+ public Map<TopicPartition, OffsetAndMetadata> complete(final Duration duration) throws ExecutionException, InterruptedException, TimeoutException {
Review Comment:
I think Kirk will fix this: We will probably abstract these kind of events into CompletableEvent
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1143865476
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -113,15 +112,45 @@ private void maybeAutoCommit(final long currentTimeMs) {
}
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptionState.allConsumed();
- log.debug("Auto-committing offsets {}", allConsumedOffsets);
sendAutoCommit(allConsumedOffsets);
autocommit.resetTimer();
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
+ * {@link UnsentOffsetCommitRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ return pendingRequests.addOffsetCommitRequest(offsets);
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ return pendingRequests.addOffsetFetchRequest(partitions);
+ }
+
+ public void clientPoll(final long currentTimeMs) {
+ this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
+ }
+
+ // Visible for testing
+ List<UnsentOffsetFetchRequest> unsentOffsetFetchRequests() {
+ return pendingRequests.unsentOffsetFetches;
+ }
+
+ // Visible for testing
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommitRequests() {
+ return pendingRequests.unsentOffsetCommits;
}
// Visible for testing
CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets) {
- CompletableFuture<ClientResponse> future = this.add(allConsumedOffsets)
+ log.debug("Enqueuing autocommit offsets: {}", allConsumedOffsets);
+ return this.addOffsetCommitRequest(allConsumedOffsets)
Review Comment:
I think you are right, we could just use the pendingQueue to handle the inflight auto commit. Thanks for spotting this.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1137588681
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
Review Comment:
I saw that we do retry on just the `COORDINATOR_LOAD_IN_PROGRESS` and `NOT_COORDINATOR` and `unstableTxnOffsetTopicPartitions`. Just confirming with you are they the only possible retriable errors for this response (saw `OffsetFetchResponse` there are others)? What about `COORDINATOR_NOT_AVAILABLE`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
+ private final UnsentOffsetFetchRequestState request;
+
+ private FetchCommittedOffsetResponseHandler(final UnsentOffsetFetchRequestState request) {
+ this.request = request;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+
+ onSuccess(currentTimeMs, response);
+ }
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ // TODO: I'm not sure if we should retry here. Sounds like we should propagate the error to let the
+ // user to fix the permission
+ request.future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ return;
+ }
+
+ private void retry(final long currentTimeMs) {
+ this.request.onFailedAttempt(currentTimeMs);
+ unsentOffsetFetchRequests.enqueue(this.request);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ request.future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ System.out.println("asdkljlsadjflksajfdlk");
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ request.future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ System.out.println("oshdofjakl");
+ // just retry
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
Review Comment:
Unlike `COORDINATOR_LOAD_IN_PROGRESS` which is a request-level global error code, `UNSTABLE_OFFSET_COMMIT` is a partition-level error. Do we always retry all the partitions as long as one of them has this error today?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
+ private final UnsentOffsetFetchRequestState request;
+
+ private FetchCommittedOffsetResponseHandler(final UnsentOffsetFetchRequestState request) {
+ this.request = request;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+
+ onSuccess(currentTimeMs, response);
+ }
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ // TODO: I'm not sure if we should retry here. Sounds like we should propagate the error to let the
+ // user to fix the permission
+ request.future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ return;
+ }
+
+ private void retry(final long currentTimeMs) {
+ this.request.onFailedAttempt(currentTimeMs);
+ unsentOffsetFetchRequests.enqueue(this.request);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ request.future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ System.out.println("asdkljlsadjflksajfdlk");
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ request.future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ System.out.println("oshdofjakl");
Review Comment:
Remove the debug line? :)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang merged pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang merged PR #13380:
URL: https://github.com/apache/kafka/pull/13380
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1140528394
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
+ private final UnsentOffsetFetchRequestState request;
+
+ private FetchCommittedOffsetResponseHandler(final UnsentOffsetFetchRequestState request) {
+ this.request = request;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+
+ onSuccess(currentTimeMs, response);
+ }
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ // TODO: I'm not sure if we should retry here. Sounds like we should propagate the error to let the
+ // user to fix the permission
+ request.future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ return;
+ }
+
+ private void retry(final long currentTimeMs) {
+ this.request.onFailedAttempt(currentTimeMs);
+ unsentOffsetFetchRequests.enqueue(this.request);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ request.future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ System.out.println("asdkljlsadjflksajfdlk");
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ request.future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ System.out.println("oshdofjakl");
+ // just retry
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
Review Comment:
Yeah, that's the current behavior. I'll leave a note here for this optimization idea.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1143881070
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
Review Comment:
The `inflightOffsetFetches` is just a temporary queue for bookkeeping purposes. So the drain does two things:
1. Return the requests that need to be send
2. Put these requests into the `inflightOffsetFetches` queue for tracking. So that we know what is currently inflight.
As you can see, there's no direct interaction with the NetworkClient in the RequestManager, so no request is actually making to the networkClient. I think the requirement is that the background thread would pool these request managers and queue up the poll results into NetworkClient directly.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1146969176
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
Review Comment:
Agreed, let's log a warning for this case.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #13380:
URL: https://github.com/apache/kafka/pull/13380#issuecomment-1482045502
Hey @guozhangwang sorry for going back and forth on this, I'll add some test cases to cover your comments.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on PR #13380:
URL: https://github.com/apache/kafka/pull/13380#issuecomment-1489456338
Made another pass, do not have further comments. LGTM
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1139257566
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
+ private final UnsentOffsetFetchRequestState request;
+
+ private FetchCommittedOffsetResponseHandler(final UnsentOffsetFetchRequestState request) {
+ this.request = request;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+
+ onSuccess(currentTimeMs, response);
+ }
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ // TODO: I'm not sure if we should retry here. Sounds like we should propagate the error to let the
+ // user to fix the permission
+ request.future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ return;
+ }
+
+ private void retry(final long currentTimeMs) {
+ this.request.onFailedAttempt(currentTimeMs);
+ unsentOffsetFetchRequests.enqueue(this.request);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ request.future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ System.out.println("asdkljlsadjflksajfdlk");
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ request.future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ System.out.println("oshdofjakl");
Review Comment:
Oh god, my secret debug weapon is revealed accidentally.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1136096815
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -80,16 +97,18 @@ public CommitRequestManager(
*/
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
- maybeAutoCommit(currentTimeMs);
+ maybeAutoCommit();
+ List<NetworkClientDelegate.UnsentRequest> unsentRequests = new ArrayList<>();
- if (stagedCommits.isEmpty()) {
- return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>());
+ if (!stagedCommits.isEmpty()) {
Review Comment:
Make sense. I think the reason I did that initially was that I wanted to send out all of the commits first before sending the fetchOffsetRequest.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1139653568
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +223,209 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
Review Comment:
I combined the handler into `UnsentOffsetFetchRequest` as we don't really need to split them
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1136449438
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -306,13 +310,28 @@ public OffsetAndMetadata committed(TopicPartition partition, Duration timeout) {
}
@Override
- public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions) {
- throw new KafkaException("method not implemented");
+ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions) {
+ return committed(partitions, Duration.ofMillis(defaultApiTimeoutMs));
}
@Override
- public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions, Duration timeout) {
- throw new KafkaException("method not implemented");
+ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions,
+ final Duration timeout) {
+ maybeThrowInvalidGroupIdException();
+ final OffsetFetchApplicationEvent event = new OffsetFetchApplicationEvent(partitions);
+ eventHandler.add(event);
+ try {
+ return event.complete(Duration.ofMillis(100));
+ } catch (ExecutionException | InterruptedException | TimeoutException e) {
+ throw new KafkaException(e);
Review Comment:
InterruptedException, TimeoutException -> We can map them to the existing kafka exception
ExecutionException - KafkaException
There are a few blocking operations like committed(), I think we need to support wakeup here. I don't know what's the suggestions here, but I'm thinking of running a timer, and check wakeup in that loop.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1139276421
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -80,16 +97,18 @@ public CommitRequestManager(
*/
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
- maybeAutoCommit(currentTimeMs);
+ maybeAutoCommit();
+ List<NetworkClientDelegate.UnsentRequest> unsentRequests = new ArrayList<>();
- if (stagedCommits.isEmpty()) {
- return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>());
+ if (!stagedCommits.isEmpty()) {
Review Comment:
One comment here, I think the intention of the RequestState is that we should retry these requests by the RequestManager instead of the NetworkClient, so this is how I think I'm going to write it:
1. `class UnsentOffsetFetchRequest extends Request State`
2. `class UnsentOffsetCommitRequest extends Request State`
3. `class UnsentRequests` holding a list of `UnsentOffsetFetchRequest` and `UnsentOffsetCommitRequest`
4. So requestManager.poll() will check a. if there are any unsent offset commits and b. if there's any sendable offsetFetch.
For inflight requests, I guess we will need to create another collection to track it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1140532145
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -149,6 +185,29 @@ Queue<StagedCommit> stagedCommits() {
return this.stagedCommits;
}
+ /**
+ * Get all the sendable requests, and create a list of UnsentRequest.
+ */
+ private List<NetworkClientDelegate.UnsentRequest> sendOffsetFetchRequests(final long currentTimeMs) {
+ List<UnsentOffsetFetchRequestState> requests = unsentOffsetFetchRequests.sendableRequests(currentTimeMs);
+ List<NetworkClientDelegate.UnsentRequest> pollResults = new ArrayList<>();
+ requests.forEach(req -> {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId, true,
+ new ArrayList<>(req.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ FetchCommittedOffsetResponseHandler cb = new FetchCommittedOffsetResponseHandler(req);
+ unsentRequest.future().whenComplete((r, t) -> {
Review Comment:
I think we need this logic to handle the disconnect. This is actually from here: https://github.com/apache/kafka/blob/3d2ac7cdbe89cdabfd95db9971de31878afa5498/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java#L605
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1143891018
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -137,31 +166,21 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
}
return null;
});
- return future;
- }
-
- public void clientPoll(final long currentTimeMs) {
- this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
}
- // Visible for testing
- Queue<StagedCommit> stagedCommits() {
- return this.stagedCommits;
- }
- private class StagedCommit {
+ private class UnsentOffsetCommitRequest {
private final Map<TopicPartition, OffsetAndMetadata> offsets;
private final String groupId;
private final GroupState.Generation generation;
private final String groupInstanceId;
private final NetworkClientDelegate.FutureCompletionHandler future;
- public StagedCommit(final Map<TopicPartition, OffsetAndMetadata> offsets,
- final String groupId,
- final String groupInstanceId,
- final GroupState.Generation generation) {
+ public UnsentOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final String groupId,
+ final String groupInstanceId,
+ final GroupState.Generation generation) {
this.offsets = offsets;
- // if no callback is provided, DefaultOffsetCommitCallback will be used.
Review Comment:
I'm moving the Callback trigger entirely to the foreground thread. As you can see, the `PrototypeAsyncConsumer` invokes:
```
future.whenComplete((r, t) -> {
if (t != null) {
callback.onComplete(offsets, new KafkaException(t));
} else {
callback.onComplete(offsets, null);
}
});
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1143888783
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
+ unsent.addAll(sendables.stream()
+ .peek(r -> r.onSendAttempt(currentTimeMs))
+ .map(r -> r.toUnsentRequest(currentTimeMs))
+ .collect(Collectors.toList()));
+ // empty the staged requests
+ unsentOffsetCommits.clear();
Review Comment:
So `unsentOffsetCommits` and `unsentOffsetFetches` tracks all of the requests that haven't been returned by the poll call (which subsequently make to the NetworkClient). So the poll calls drain() in the end, and dumping the unsentRequest into the `inflightOffsetFetches`.
I think the naming and the way the code is structured is probably too confusing. Maybe there's a better way to bookkeep these inflight/unsent requests.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on PR #13380:
URL: https://github.com/apache/kafka/pull/13380#issuecomment-1474361224
@philipnee let me know when it's ready for another look.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1143739168
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
Review Comment:
yap that sounds good!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1139650990
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -137,31 +144,40 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
}
return null;
});
- return future;
}
- public void clientPoll(final long currentTimeMs) {
- this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
+ * {@link UnsentOffsetCommitRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
Review Comment:
just to make both method return type and naming consistent.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1140442417
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -149,6 +185,29 @@ Queue<StagedCommit> stagedCommits() {
return this.stagedCommits;
}
+ /**
+ * Get all the sendable requests, and create a list of UnsentRequest.
+ */
+ private List<NetworkClientDelegate.UnsentRequest> sendOffsetFetchRequests(final long currentTimeMs) {
+ List<UnsentOffsetFetchRequestState> requests = unsentOffsetFetchRequests.sendableRequests(currentTimeMs);
+ List<NetworkClientDelegate.UnsentRequest> pollResults = new ArrayList<>();
+ requests.forEach(req -> {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId, true,
+ new ArrayList<>(req.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ FetchCommittedOffsetResponseHandler cb = new FetchCommittedOffsetResponseHandler(req);
Review Comment:
I merged it with the offset fetch class.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1146961554
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
+ unsent.addAll(sendables.stream()
+ .peek(r -> r.onSendAttempt(currentTimeMs))
+ .map(r -> r.toUnsentRequest(currentTimeMs))
+ .collect(Collectors.toList()));
+ // empty the staged requests
+ unsentOffsetCommits.clear();
Review Comment:
Yeah, I think we should not just blindly empty the `unsentOffsetFetches`, instead we should only remove `sendables` from the queue. Could you fix it?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1136454609
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
Review Comment:
It's 80% copy paste, a thing changed here:
`retry(currentTimeMs);` is invoked upon RetriableExceptions. In the original code, the exception was raised and then retried until timeout. Here is the reference: https://github.com/apache/kafka/blob/d23ce20bdfbe5a9598523961cb7cf747ce4f52ef/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java#L1016
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1133441978
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -149,6 +185,29 @@ Queue<StagedCommit> stagedCommits() {
return this.stagedCommits;
}
+ /**
+ * Get all the sendable requests, and create a list of UnsentRequest.
+ */
+ private List<NetworkClientDelegate.UnsentRequest> sendOffsetFetchRequests(final long currentTimeMs) {
+ List<UnsentOffsetFetchRequestState> requests = unsentOffsetFetchRequests.sendableRequests(currentTimeMs);
+ List<NetworkClientDelegate.UnsentRequest> pollResults = new ArrayList<>();
+ requests.forEach(req -> {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId, true,
+ new ArrayList<>(req.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ FetchCommittedOffsetResponseHandler cb = new FetchCommittedOffsetResponseHandler(req);
Review Comment:
I think this can be cleaner, so please give me some suggestions here!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1146971049
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -137,31 +166,21 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
}
return null;
});
- return future;
- }
-
- public void clientPoll(final long currentTimeMs) {
- this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
}
- // Visible for testing
- Queue<StagedCommit> stagedCommits() {
- return this.stagedCommits;
- }
- private class StagedCommit {
+ private class UnsentOffsetCommitRequest {
private final Map<TopicPartition, OffsetAndMetadata> offsets;
private final String groupId;
private final GroupState.Generation generation;
private final String groupInstanceId;
private final NetworkClientDelegate.FutureCompletionHandler future;
- public StagedCommit(final Map<TopicPartition, OffsetAndMetadata> offsets,
- final String groupId,
- final String groupInstanceId,
- final GroupState.Generation generation) {
+ public UnsentOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final String groupId,
+ final String groupInstanceId,
+ final GroupState.Generation generation) {
this.offsets = offsets;
- // if no callback is provided, DefaultOffsetCommitCallback will be used.
Review Comment:
Thanks, I'll add a test case for this.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1143999124
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -137,31 +166,21 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
}
return null;
});
- return future;
- }
-
- public void clientPoll(final long currentTimeMs) {
- this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
}
- // Visible for testing
- Queue<StagedCommit> stagedCommits() {
- return this.stagedCommits;
- }
- private class StagedCommit {
+ private class UnsentOffsetCommitRequest {
private final Map<TopicPartition, OffsetAndMetadata> offsets;
private final String groupId;
private final GroupState.Generation generation;
private final String groupInstanceId;
private final NetworkClientDelegate.FutureCompletionHandler future;
- public StagedCommit(final Map<TopicPartition, OffsetAndMetadata> offsets,
- final String groupId,
- final String groupInstanceId,
- final GroupState.Generation generation) {
+ public UnsentOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final String groupId,
+ final String groupInstanceId,
+ final GroupState.Generation generation) {
this.offsets = offsets;
- // if no callback is provided, DefaultOffsetCommitCallback will be used.
Review Comment:
I'll double check if the defaultOffsetCommitCallback is being invoked when user provides a null callback.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1146959499
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
Review Comment:
I was actually thinking that if it is not expected for normal cases, maybe we should log it as WARN instead, or even throw a fatal error if we believe it should never happen.
For the case you described: in that case, the `drain` should not return any unsent request to the network client, and hence their requests would not be sent, and hence no responses should be received triggering this logic, right? So that seems still not normally happening?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1137547234
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +266,85 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ static class UnsentOffsetFetchRequestState extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequestState(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = future;
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequestState request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+ }
+
+ /**
+ * <p>This is used to support the committed() API. Here we use a Java Collections, {@code unsentRequests}, to
+ * track
+ * the OffsetFetchRequests that haven't been sent, to prevent sending the same requests in the same batch.
+ *
+ * <p>If the request is new. It will be enqueued to the {@code unsentRequest}, and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * TODO: There is an optimization to present duplication to the sent but incompleted requests. I'm not sure if we
+ * need that.
+ */
+ class UnsentOffsetFetchRequests {
Review Comment:
Maybe it's not that common; but the current flag we used is to cover both periods, and I wonder if was conducted that way to avoid some surprising cases indeed.. So my rationale is just to keep the behavior consistent with the old code logic.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1136315295
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +266,85 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ static class UnsentOffsetFetchRequestState extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequestState(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = future;
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequestState request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+ }
+
+ /**
+ * <p>This is used to support the committed() API. Here we use a Java Collections, {@code unsentRequests}, to
+ * track
+ * the OffsetFetchRequests that haven't been sent, to prevent sending the same requests in the same batch.
+ *
+ * <p>If the request is new. It will be enqueued to the {@code unsentRequest}, and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * TODO: There is an optimization to present duplication to the sent but incompleted requests. I'm not sure if we
+ * need that.
+ */
+ class UnsentOffsetFetchRequests {
Review Comment:
Just to think a bit further here, it makes me thinking that inside the `unsentOffsetFetchRequests` we'd need to keep two collections: 1) the unsent requests, 2) the sent-but-not-responded requests, and upon getting a new event from the queue, we would check against both collections. And then we we drain the first collection and write them to the network client, we move them to the second collection, and only drop requests from the second collection after a response is received and the registered handlers (it's possible to have multiple events' handlers registered for the same request).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1140667236
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
+ private final UnsentOffsetFetchRequestState request;
+
+ private FetchCommittedOffsetResponseHandler(final UnsentOffsetFetchRequestState request) {
+ this.request = request;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+
+ onSuccess(currentTimeMs, response);
+ }
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ // TODO: I'm not sure if we should retry here. Sounds like we should propagate the error to let the
+ // user to fix the permission
+ request.future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ return;
+ }
+
+ private void retry(final long currentTimeMs) {
+ this.request.onFailedAttempt(currentTimeMs);
+ unsentOffsetFetchRequests.enqueue(this.request);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ request.future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ System.out.println("asdkljlsadjflksajfdlk");
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ request.future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ request.future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ System.out.println("oshdofjakl");
+ // just retry
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
Review Comment:
Sounds good! Thanks for confirming.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #13380:
URL: https://github.com/apache/kafka/pull/13380#issuecomment-1474404071
Hey @guozhangwang - I think i've addressed most of your questions except merging the request classes. It is a bit difficult right now as we sometimes need to access the fields before `UnsentRequest` is created.
I've also updated the dedupcation before adding the commit - There are tests verifying the requests actually got purged upon completion
I changed the naming of the methods and classes to make things more consistent.
I've combined the request handler with the offset fetch request to make things cleaner.
The Java exceptions, per your suggestions, are mapped to kafka exception. So the same set of exception will be thrown
TODO:
- Need to implement wakeup
- Abstract some of ApplicationEvent to maybe something like CompletableEvent as some of them have a completableFuture embedded in it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1146967451
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
+ unsent.addAll(sendables.stream()
+ .peek(r -> r.onSendAttempt(currentTimeMs))
+ .map(r -> r.toUnsentRequest(currentTimeMs))
+ .collect(Collectors.toList()));
+ // empty the staged requests
+ unsentOffsetCommits.clear();
Review Comment:
The following line should address clearing the unsendables I think?
i.e.:
`unsentOffsetFetches.addAll(partitionedBySendability.get(false));`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on PR #13380:
URL: https://github.com/apache/kafka/pull/13380#issuecomment-1482028180
@philipnee I made a third pass on the PR, but it seems we still have a few issues not fixed: 1) https://github.com/apache/kafka/pull/13380#discussion_r1146961554, 2) https://github.com/apache/kafka/pull/13380#discussion_r1146963166, and 3) https://github.com/apache/kafka/pull/13380/files/d5013568e590768ccce6601bb271ff6a50bc1299#r1142698917. LMK what do you think?
If you agree they should be fixed, while you fix them could you please also add the corresponding test cases to cover those scenarios?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1148140557
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -113,15 +112,45 @@ private void maybeAutoCommit(final long currentTimeMs) {
}
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptionState.allConsumed();
- log.debug("Auto-committing offsets {}", allConsumedOffsets);
sendAutoCommit(allConsumedOffsets);
autocommit.resetTimer();
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
+ * {@link UnsentOffsetCommitRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ return pendingRequests.addOffsetCommitRequest(offsets);
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ return pendingRequests.addOffsetFetchRequest(partitions);
+ }
+
+ public void clientPoll(final long currentTimeMs) {
+ this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
+ }
+
+ // Visible for testing
+ List<UnsentOffsetFetchRequest> unsentOffsetFetchRequests() {
+ return pendingRequests.unsentOffsetFetches;
+ }
+
+ // Visible for testing
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommitRequests() {
+ return pendingRequests.unsentOffsetCommits;
}
// Visible for testing
CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets) {
- CompletableFuture<ClientResponse> future = this.add(allConsumedOffsets)
+ log.debug("Enqueuing autocommit offsets: {}", allConsumedOffsets);
+ return this.addOffsetCommitRequest(allConsumedOffsets)
Review Comment:
Hey I somehow thought I addressed this comment by setting the `hasInflightCommit` but I somehow can't find it here. I'll follow this up with a commit - along with testing the commit callback. Sorry about that.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1134383599
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +266,85 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ static class UnsentOffsetFetchRequestState extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequestState(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = future;
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequestState request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+ }
+
+ /**
+ * <p>This is used to support the committed() API. Here we use a Java Collections, {@code unsentRequests}, to
+ * track
+ * the OffsetFetchRequests that haven't been sent, to prevent sending the same requests in the same batch.
+ *
+ * <p>If the request is new. It will be enqueued to the {@code unsentRequest}, and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * TODO: There is an optimization to present duplication to the sent but incompleted requests. I'm not sure if we
+ * need that.
+ */
+ class UnsentOffsetFetchRequests {
+ private List<UnsentOffsetFetchRequestState> unsentRequests = new ArrayList<>();
+
+ public boolean canSendOffsetFetch() {
Review Comment:
I think the func name could be misleading in the long run. Let's just use more specific name like `hasUnsentRequests`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -140,6 +158,24 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
return future;
}
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequestState} and enqueue it to send later.
+ */
+ public void addCommittedRequest(final Set<TopicPartition> partitions,
Review Comment:
nit: let's rename the other `add` func to `addCommitOffsetRequest` to be more distinguishable with this one?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +266,85 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ static class UnsentOffsetFetchRequestState extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequestState(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = future;
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequestState request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+ }
+
+ /**
+ * <p>This is used to support the committed() API. Here we use a Java Collections, {@code unsentRequests}, to
+ * track
+ * the OffsetFetchRequests that haven't been sent, to prevent sending the same requests in the same batch.
+ *
+ * <p>If the request is new. It will be enqueued to the {@code unsentRequest}, and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * TODO: There is an optimization to present duplication to the sent but incompleted requests. I'm not sure if we
+ * need that.
+ */
+ class UnsentOffsetFetchRequests {
Review Comment:
The original flag is used not only to cover the period when there's already a request not yet sent, but also during the time when the request is sent but the response is not yet received. It seems that we would not cover the second period i.e. if a request is sent and response not yet received, we may send the same request again.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -149,6 +185,29 @@ Queue<StagedCommit> stagedCommits() {
return this.stagedCommits;
}
+ /**
+ * Get all the sendable requests, and create a list of UnsentRequest.
+ */
+ private List<NetworkClientDelegate.UnsentRequest> sendOffsetFetchRequests(final long currentTimeMs) {
Review Comment:
Again the func name is a bit misleading. What about `constructOffsetFetchUnsentRequests`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -140,6 +158,24 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
return future;
}
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequestState} and enqueue it to send later.
+ */
+ public void addCommittedRequest(final Set<TopicPartition> partitions,
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ if (partitions.isEmpty()) {
+ future.complete(new HashMap<>());
+ return;
+ }
+
+ unsentOffsetFetchRequests.enqueue(new UnsentOffsetFetchRequestState(
Review Comment:
This one reminds me: if we can directly construct the `RequestState` here, we we cannot do the same for commit request and instead have to add to a `stagedCommit` first?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -149,6 +185,29 @@ Queue<StagedCommit> stagedCommits() {
return this.stagedCommits;
}
+ /**
+ * Get all the sendable requests, and create a list of UnsentRequest.
+ */
+ private List<NetworkClientDelegate.UnsentRequest> sendOffsetFetchRequests(final long currentTimeMs) {
+ List<UnsentOffsetFetchRequestState> requests = unsentOffsetFetchRequests.sendableRequests(currentTimeMs);
+ List<NetworkClientDelegate.UnsentRequest> pollResults = new ArrayList<>();
+ requests.forEach(req -> {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId, true,
+ new ArrayList<>(req.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ FetchCommittedOffsetResponseHandler cb = new FetchCommittedOffsetResponseHandler(req);
+ unsentRequest.future().whenComplete((r, t) -> {
Review Comment:
Why we use a `FetchCommittedOffsetResponseHandler` for committed API, while using a general `FutureCompletionHandler` for commit API? Is this inconsistency intentional?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -306,13 +310,28 @@ public OffsetAndMetadata committed(TopicPartition partition, Duration timeout) {
}
@Override
- public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions) {
- throw new KafkaException("method not implemented");
+ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions) {
+ return committed(partitions, Duration.ofMillis(defaultApiTimeoutMs));
}
@Override
- public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions, Duration timeout) {
- throw new KafkaException("method not implemented");
+ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions,
+ final Duration timeout) {
+ maybeThrowInvalidGroupIdException();
+ final OffsetFetchApplicationEvent event = new OffsetFetchApplicationEvent(partitions);
+ eventHandler.add(event);
+ try {
+ return event.complete(Duration.ofMillis(100));
+ } catch (ExecutionException | InterruptedException | TimeoutException e) {
+ throw new KafkaException(e);
Review Comment:
I think it's not... why not read through the current code and make the throwing behavior more consistent?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -80,16 +97,18 @@ public CommitRequestManager(
*/
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
- maybeAutoCommit(currentTimeMs);
+ maybeAutoCommit();
+ List<NetworkClientDelegate.UnsentRequest> unsentRequests = new ArrayList<>();
- if (stagedCommits.isEmpty()) {
- return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>());
+ if (!stagedCommits.isEmpty()) {
Review Comment:
I think with the new `unsentOffsetFetchRequests` the `stagedCommits` can be merged into it, since it seems unnecessary to keep two sets of unsent requests just to pour one of its content into another periodically.
Also I feel that it's not necessarily to keep a second placeholder for `UnsentRequest` in `UnsentOffsetFetchRequestState`.
More specifically, upon `manager.add` we can check whether there's a dup and if not construct an `UnsentRequest` with the handler hooked right away and add to the `unsentOffsetFetchRequests`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -140,6 +158,24 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
return future;
}
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequestState} and enqueue it to send later.
+ */
+ public void addCommittedRequest(final Set<TopicPartition> partitions,
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ if (partitions.isEmpty()) {
+ future.complete(new HashMap<>());
+ return;
+ }
+
+ unsentOffsetFetchRequests.enqueue(new UnsentOffsetFetchRequestState(
Review Comment:
Okay this train of thought leads me to suggest merging the two as well as merging the `UnsentOffsetFetchRequestState` with `UnsentRequest` as well (see other detailed comments). LMK what do you think.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -80,16 +97,18 @@ public CommitRequestManager(
*/
@Override
public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
- maybeAutoCommit(currentTimeMs);
+ maybeAutoCommit();
+ List<NetworkClientDelegate.UnsentRequest> unsentRequests = new ArrayList<>();
- if (stagedCommits.isEmpty()) {
- return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>());
+ if (!stagedCommits.isEmpty()) {
Review Comment:
Of course that means we need the `UnsentRequest` to also include a field of `RequestState` to be able to check `canSendRequest` and mark attempt timestamp etc.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +266,85 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ static class UnsentOffsetFetchRequestState extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequestState(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = future;
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequestState request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+ }
+
+ /**
+ * <p>This is used to support the committed() API. Here we use a Java Collections, {@code unsentRequests}, to
+ * track
+ * the OffsetFetchRequests that haven't been sent, to prevent sending the same requests in the same batch.
+ *
+ * <p>If the request is new. It will be enqueued to the {@code unsentRequest}, and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * TODO: There is an optimization to present duplication to the sent but incompleted requests. I'm not sure if we
Review Comment:
Yeah I think we can defer that for future work.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
Review Comment:
Assuming this is just copy-paste here, without any handler logic changes, so I skipped this class. @philipnee let me know if otherwise.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -149,6 +185,29 @@ Queue<StagedCommit> stagedCommits() {
return this.stagedCommits;
}
+ /**
+ * Get all the sendable requests, and create a list of UnsentRequest.
+ */
+ private List<NetworkClientDelegate.UnsentRequest> sendOffsetFetchRequests(final long currentTimeMs) {
+ List<UnsentOffsetFetchRequestState> requests = unsentOffsetFetchRequests.sendableRequests(currentTimeMs);
+ List<NetworkClientDelegate.UnsentRequest> pollResults = new ArrayList<>();
+ requests.forEach(req -> {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId, true,
+ new ArrayList<>(req.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ FetchCommittedOffsetResponseHandler cb = new FetchCommittedOffsetResponseHandler(req);
Review Comment:
I think it is actually fine :) But just to echo my other comments, I think we can merge `stagedCommits` with this class as well as eliminating `UnsentOffsetFetchRequestState`, in which case we will always construct the `UnsentRequest` right away and add to the list, so that we do not need to do it here.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -437,29 +456,24 @@ public void wakeup() {
*/
@Override
public void commitSync(final Duration timeout) {
- final CommitApplicationEvent commitEvent = new CommitApplicationEvent(subscriptions.allConsumed());
- eventHandler.add(commitEvent);
-
- final CompletableFuture<Void> commitFuture = commitEvent.future();
- try {
- commitFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
- } catch (final TimeoutException e) {
- throw new org.apache.kafka.common.errors.TimeoutException(
- "timeout");
- } catch (final Exception e) {
- // handle exception here
- throw new RuntimeException(e);
- }
+ commitSync(subscriptions.allConsumed(), timeout);
}
@Override
public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets) {
- throw new KafkaException("method not implemented");
+ commitSync(offsets, Duration.ofMillis(defaultApiTimeoutMs));
}
@Override
public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets, Duration timeout) {
- throw new KafkaException("method not implemented");
+ CompletableFuture<Void> commitFuture = commit(offsets);
+ try {
+ commitFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (final TimeoutException e) {
Review Comment:
Ditto here.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1136104123
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -149,6 +185,29 @@ Queue<StagedCommit> stagedCommits() {
return this.stagedCommits;
}
+ /**
+ * Get all the sendable requests, and create a list of UnsentRequest.
+ */
+ private List<NetworkClientDelegate.UnsentRequest> sendOffsetFetchRequests(final long currentTimeMs) {
+ List<UnsentOffsetFetchRequestState> requests = unsentOffsetFetchRequests.sendableRequests(currentTimeMs);
+ List<NetworkClientDelegate.UnsentRequest> pollResults = new ArrayList<>();
+ requests.forEach(req -> {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId, true,
+ new ArrayList<>(req.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ FetchCommittedOffsetResponseHandler cb = new FetchCommittedOffsetResponseHandler(req);
+ unsentRequest.future().whenComplete((r, t) -> {
Review Comment:
When we first made the FutureCompletionHandler, we explicitly handled the responses in the onComplete method here:
``` if (response.authenticationException() != null) {
onFailure(response.authenticationException());
} else if (response.wasDisconnected()) {
onFailure(DisconnectException.INSTANCE);
} else if (response.versionMismatch() != null) {
onFailure(response.versionMismatch());
} else {
future.complete(response);
}
```
I was thinking about your suggestion, but since you mentioned it again, let me try to refactor this a bit to make it more pluggin-able
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1140429707
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -116,12 +114,21 @@ private void maybeAutoCommit(final long currentTimeMs) {
log.debug("Auto-committing offsets {}", allConsumedOffsets);
Review Comment:
I feel this message is misleading - but in the new model, we batch the commits and send them out at the network client poll in the background thread. Maybe we word it as "enqueuing autocommit offsets {}", offsets ?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1136454999
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -437,29 +456,24 @@ public void wakeup() {
*/
@Override
public void commitSync(final Duration timeout) {
- final CommitApplicationEvent commitEvent = new CommitApplicationEvent(subscriptions.allConsumed());
- eventHandler.add(commitEvent);
-
- final CompletableFuture<Void> commitFuture = commitEvent.future();
- try {
- commitFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
- } catch (final TimeoutException e) {
- throw new org.apache.kafka.common.errors.TimeoutException(
- "timeout");
- } catch (final Exception e) {
- // handle exception here
- throw new RuntimeException(e);
- }
+ commitSync(subscriptions.allConsumed(), timeout);
}
@Override
public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets) {
- throw new KafkaException("method not implemented");
+ commitSync(offsets, Duration.ofMillis(defaultApiTimeoutMs));
}
@Override
public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets, Duration timeout) {
- throw new KafkaException("method not implemented");
+ CompletableFuture<Void> commitFuture = commit(offsets);
+ try {
+ commitFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (final TimeoutException e) {
Review Comment:
Ditto above: InterruptedException -> kafka interruptException. TimeoutException -> kafka timeoutException. ExecutionException -> kafkaException
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1136434874
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +266,85 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ static class UnsentOffsetFetchRequestState extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequestState(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = future;
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequestState request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+ }
+
+ /**
+ * <p>This is used to support the committed() API. Here we use a Java Collections, {@code unsentRequests}, to
+ * track
+ * the OffsetFetchRequests that haven't been sent, to prevent sending the same requests in the same batch.
+ *
+ * <p>If the request is new. It will be enqueued to the {@code unsentRequest}, and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * TODO: There is an optimization to present duplication to the sent but incompleted requests. I'm not sure if we
+ * need that.
+ */
+ class UnsentOffsetFetchRequests {
Review Comment:
Hi! That was my original attempt :) , but I abandon it because I was wondering if we actually need this optimization?
I think if we call committed() several times consecutively within a single event loop, it makes sense to coalesce them into a single request. Once we send out the request, I wonder how often do we invoke another committed() before the request gets sent out. Is it a use case from the stream side?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1133345989
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -306,13 +310,28 @@ public OffsetAndMetadata committed(TopicPartition partition, Duration timeout) {
}
@Override
- public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions) {
- throw new KafkaException("method not implemented");
+ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions) {
+ return committed(partitions, Duration.ofMillis(defaultApiTimeoutMs));
}
@Override
- public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions, Duration timeout) {
- throw new KafkaException("method not implemented");
+ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions,
+ final Duration timeout) {
+ maybeThrowInvalidGroupIdException();
+ final OffsetFetchApplicationEvent event = new OffsetFetchApplicationEvent(partitions);
+ eventHandler.add(event);
+ try {
+ return event.complete(Duration.ofMillis(100));
+ } catch (ExecutionException | InterruptedException | TimeoutException e) {
+ throw new KafkaException(e);
Review Comment:
I actually don't know if this is right.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1142683410
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
Review Comment:
nit: we can also leave a TODO comment here to de-dup offset commit requests if they are committing the exactly same offset for those partitions?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
Review Comment:
How about renaming to `OffsetFetchRequestState`, and ditto for `OffsetCommitRequestState`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
Review Comment:
Just curious when do you think this could ever happen?
One case I was thinking is if the consumer is shutting down, but I also feel that when we shutdown a consumer, we should try to stop the background thread first, and then clear the buffer, so that the response would not be received while the buffer is cleared.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -113,15 +112,45 @@ private void maybeAutoCommit(final long currentTimeMs) {
}
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptionState.allConsumed();
- log.debug("Auto-committing offsets {}", allConsumedOffsets);
sendAutoCommit(allConsumedOffsets);
autocommit.resetTimer();
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
+ * {@link UnsentOffsetCommitRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ return pendingRequests.addOffsetCommitRequest(offsets);
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ return pendingRequests.addOffsetFetchRequest(partitions);
+ }
+
+ public void clientPoll(final long currentTimeMs) {
+ this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
+ }
+
+ // Visible for testing
+ List<UnsentOffsetFetchRequest> unsentOffsetFetchRequests() {
+ return pendingRequests.unsentOffsetFetches;
+ }
+
+ // Visible for testing
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommitRequests() {
+ return pendingRequests.unsentOffsetCommits;
}
// Visible for testing
CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets) {
- CompletableFuture<ClientResponse> future = this.add(allConsumedOffsets)
+ log.debug("Enqueuing autocommit offsets: {}", allConsumedOffsets);
+ return this.addOffsetCommitRequest(allConsumedOffsets)
Review Comment:
I have a question for line 159 below: it seems we only `setInflightCommitStatus(false)`, but never `setInflightCommitStatus(true)`. That means `hasInflightCommit` would always be false? Or did I miss anything?
Also I feel that with the current `pending request` queue, we may not need the `hasInflightCommit` flag any more --- as long as there are commit requests in the pending queue that covers all the partitions in interest, we can omit sending that again, right?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
+ unsent.addAll(sendables.stream()
+ .peek(r -> r.onSendAttempt(currentTimeMs))
+ .map(r -> r.toUnsentRequest(currentTimeMs))
+ .collect(Collectors.toList()));
+ // empty the staged requests
Review Comment:
This comment seems redundant and can be removed, as it is pretty clear from the code. Also the term `staged` maybe confusing as we are now in `unsent` vocabulary.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -113,15 +112,45 @@ private void maybeAutoCommit(final long currentTimeMs) {
}
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptionState.allConsumed();
- log.debug("Auto-committing offsets {}", allConsumedOffsets);
sendAutoCommit(allConsumedOffsets);
autocommit.resetTimer();
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
+ * {@link UnsentOffsetCommitRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ return pendingRequests.addOffsetCommitRequest(offsets);
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ return pendingRequests.addOffsetFetchRequest(partitions);
+ }
+
+ public void clientPoll(final long currentTimeMs) {
Review Comment:
The func name `clientPoll` and `poll` are a bit less distinguishable. It seems the former function would only be used to update the autoCommit state of the commitManager, right? If we do not expect this function to be extended for other logic, could we just rename it to `updateAutoCommitTimer`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
+ unsent.addAll(sendables.stream()
+ .peek(r -> r.onSendAttempt(currentTimeMs))
+ .map(r -> r.toUnsentRequest(currentTimeMs))
+ .collect(Collectors.toList()));
+ // empty the staged requests
+ unsentOffsetCommits.clear();
Review Comment:
This seems not right to me as well: if an unsent request is filtered in `.filter(r -> r.canSendRequest(currentTimeMs))` they should still be kept in the queue, right?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -137,31 +166,21 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
}
return null;
});
- return future;
- }
-
- public void clientPoll(final long currentTimeMs) {
- this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
}
- // Visible for testing
- Queue<StagedCommit> stagedCommits() {
- return this.stagedCommits;
- }
- private class StagedCommit {
+ private class UnsentOffsetCommitRequest {
private final Map<TopicPartition, OffsetAndMetadata> offsets;
private final String groupId;
private final GroupState.Generation generation;
private final String groupInstanceId;
private final NetworkClientDelegate.FutureCompletionHandler future;
- public StagedCommit(final Map<TopicPartition, OffsetAndMetadata> offsets,
- final String groupId,
- final String groupInstanceId,
- final GroupState.Generation generation) {
+ public UnsentOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final String groupId,
+ final String groupInstanceId,
+ final GroupState.Generation generation) {
this.offsets = offsets;
- // if no callback is provided, DefaultOffsetCommitCallback will be used.
Review Comment:
I'm wondering in the new code, did we still trigger `DefaultOffsetCommitCallback` if user does not specify a callback? Currently it seems we always just use a plain `FutureCompletionHandler.future` without registering any callback. Is that a behavioral change?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +223,209 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
Review Comment:
Not sure I understand this comment?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
Review Comment:
Ack, let's keep the current behavior for now.
Maybe we can add a TODO here to consider retry for that error in the future as a behavioral change?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
Review Comment:
This seems not right to me: if we add the inflight request to sendables again and return in `drain`, then they will be re-sent by the network client delegate?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #13380:
URL: https://github.com/apache/kafka/pull/13380#issuecomment-1487707392
There's this weird failing test that I can't resolve. I think I might need to rebase.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1143738304
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
Review Comment:
Probably unlikely - unless `drain()` got interrupted before dumping the requests to the inflight queue. Would you prefer to remove this log?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1146963166
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -137,31 +166,21 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
}
return null;
});
- return future;
- }
-
- public void clientPoll(final long currentTimeMs) {
- this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
}
- // Visible for testing
- Queue<StagedCommit> stagedCommits() {
- return this.stagedCommits;
- }
- private class StagedCommit {
+ private class UnsentOffsetCommitRequest {
private final Map<TopicPartition, OffsetAndMetadata> offsets;
private final String groupId;
private final GroupState.Generation generation;
private final String groupInstanceId;
private final NetworkClientDelegate.FutureCompletionHandler future;
- public StagedCommit(final Map<TopicPartition, OffsetAndMetadata> offsets,
- final String groupId,
- final String groupInstanceId,
- final GroupState.Generation generation) {
+ public UnsentOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final String groupId,
+ final String groupInstanceId,
+ final GroupState.Generation generation) {
this.offsets = offsets;
- // if no callback is provided, DefaultOffsetCommitCallback will be used.
Review Comment:
I do not think we do: in case `commitAync()` is called we just passed in `null` as the cb. In that case we would actually have an NPE when triggering:
```
callback.onComplete(offsets, new RuntimeException(t));
```
right?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1146979021
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -137,31 +166,21 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
}
return null;
});
- return future;
- }
-
- public void clientPoll(final long currentTimeMs) {
- this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
}
- // Visible for testing
- Queue<StagedCommit> stagedCommits() {
- return this.stagedCommits;
- }
- private class StagedCommit {
+ private class UnsentOffsetCommitRequest {
private final Map<TopicPartition, OffsetAndMetadata> offsets;
private final String groupId;
private final GroupState.Generation generation;
private final String groupInstanceId;
private final NetworkClientDelegate.FutureCompletionHandler future;
- public StagedCommit(final Map<TopicPartition, OffsetAndMetadata> offsets,
- final String groupId,
- final String groupInstanceId,
- final GroupState.Generation generation) {
+ public UnsentOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final String groupId,
+ final String groupInstanceId,
+ final GroupState.Generation generation) {
this.offsets = offsets;
- // if no callback is provided, DefaultOffsetCommitCallback will be used.
Review Comment:
I believe this was addressed in the last revision. See PrototypeAsyncConsumer:
```
final OffsetCommitCallback commitCallback = callback == null ? new DefaultOffsetCommitCallback() : callback;
future.whenComplete((r, t) -> { ... } });
```
But i'll add a test for this.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1144002857
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
+ unsent.addAll(sendables.stream()
+ .peek(r -> r.onSendAttempt(currentTimeMs))
+ .map(r -> r.toUnsentRequest(currentTimeMs))
+ .collect(Collectors.toList()));
+ // empty the staged requests
+ unsentOffsetCommits.clear();
Review Comment:
seems like a bug here: I cleared the non-sendables.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1148127234
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -113,15 +112,45 @@ private void maybeAutoCommit(final long currentTimeMs) {
}
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptionState.allConsumed();
- log.debug("Auto-committing offsets {}", allConsumedOffsets);
sendAutoCommit(allConsumedOffsets);
autocommit.resetTimer();
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
+ * {@link UnsentOffsetCommitRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ return pendingRequests.addOffsetCommitRequest(offsets);
+ }
+
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ return pendingRequests.addOffsetFetchRequest(partitions);
+ }
+
+ public void clientPoll(final long currentTimeMs) {
+ this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
+ }
+
+ // Visible for testing
+ List<UnsentOffsetFetchRequest> unsentOffsetFetchRequests() {
+ return pendingRequests.unsentOffsetFetches;
+ }
+
+ // Visible for testing
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommitRequests() {
+ return pendingRequests.unsentOffsetCommits;
}
// Visible for testing
CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets) {
- CompletableFuture<ClientResponse> future = this.add(allConsumedOffsets)
+ log.debug("Enqueuing autocommit offsets: {}", allConsumedOffsets);
+ return this.addOffsetCommitRequest(allConsumedOffsets)
Review Comment:
What do you think about this comment so far?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -207,6 +226,232 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
}
}
+ private class UnsentOffsetFetchRequest extends RequestState {
+ public final Set<TopicPartition> requestedPartitions;
+ public final GroupState.Generation requestedGeneration;
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ public UnsentOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final GroupState.Generation generation,
+ final long retryBackoffMs) {
+ super(retryBackoffMs);
+ this.requestedPartitions = partitions;
+ this.requestedGeneration = generation;
+ this.future = new CompletableFuture<>();
+ }
+
+ public boolean sameRequest(final UnsentOffsetFetchRequest request) {
+ return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ }
+
+ public NetworkClientDelegate.UnsentRequest toUnsentRequest(final long currentTimeMs) {
+ OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
+ groupState.groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ NetworkClientDelegate.UnsentRequest unsentRequest = new NetworkClientDelegate.UnsentRequest(
+ builder,
+ coordinatorRequestManager.coordinator());
+ unsentRequest.future().whenComplete((r, t) -> {
+ onResponse(currentTimeMs, (OffsetFetchResponse) r.responseBody());
+ });
+ return unsentRequest;
+ }
+
+ public void onResponse(
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupState.groupId);
+ if (responseError != Errors.NONE) {
+ onFailure(currentTimeMs, responseError);
+ return;
+ }
+ onSuccess(currentTimeMs, response);
+ }
+
+ private void onFailure(final long currentTimeMs,
+ final Errors responseError) {
+ log.debug("Offset fetch failed: {}", responseError.message());
+
+ // TODO: should we retry on COORDINATOR_NOT_AVAILABLE as well ?
+ if (responseError == Errors.COORDINATOR_LOAD_IN_PROGRESS) {
+ retry(currentTimeMs);
+ } else if (responseError == Errors.NOT_COORDINATOR) {
+ // re-discover the coordinator and retry
+ coordinatorRequestManager.markCoordinatorUnknown(responseError.message(), Time.SYSTEM.milliseconds());
+ retry(currentTimeMs);
+ } else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
+ }
+ }
+
+ private void retry(final long currentTimeMs) {
+ onFailedAttempt(currentTimeMs);
+ onSendAttempt(currentTimeMs);
+ pendingRequests.addOffsetFetchRequest(this);
+ }
+
+ private void onSuccess(final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Set<String> unauthorizedTopics = null;
+ Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
+ response.partitionDataMap(groupState.groupId);
+ Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
+ Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
+ for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
+ TopicPartition tp = entry.getKey();
+ OffsetFetchResponse.PartitionData partitionData = entry.getValue();
+ if (partitionData.hasError()) {
+ Errors error = partitionData.error;
+ log.debug("Failed to fetch offset for partition {}: {}", tp, error.message());
+
+ if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ future.completeExceptionally(new KafkaException("Topic or Partition " + tp + " does " +
+ "not " +
+ "exist"));
+ return;
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (unauthorizedTopics == null) {
+ unauthorizedTopics = new HashSet<>();
+ }
+ unauthorizedTopics.add(tp.topic());
+ } else if (error == Errors.UNSTABLE_OFFSET_COMMIT) {
+ unstableTxnOffsetTopicPartitions.add(tp);
+ } else {
+ future.completeExceptionally(new KafkaException("Unexpected error in fetch offset " +
+ "response for partition " + tp + ": " + error.message()));
+ return;
+ }
+ } else if (partitionData.offset >= 0) {
+ // record the position with the offset (-1 indicates no committed offset to fetch);
+ // if there's no committed offset, record as null
+ offsets.put(tp, new OffsetAndMetadata(partitionData.offset, partitionData.leaderEpoch, partitionData.metadata));
+ } else {
+ log.info("Found no committed offset for partition {}", tp);
+ offsets.put(tp, null);
+ }
+ }
+
+ if (unauthorizedTopics != null) {
+ future.completeExceptionally(new TopicAuthorizationException(unauthorizedTopics));
+ } else if (!unstableTxnOffsetTopicPartitions.isEmpty()) {
+ // TODO: Optimization question: Do we need to retry all partitions upon a single partition error?
+ log.info("The following partitions still have unstable offsets " +
+ "which are not cleared on the broker side: {}" +
+ ", this could be either " +
+ "transactional offsets waiting for completion, or " +
+ "normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
+ retry(currentTimeMs);
+ } else {
+ future.complete(offsets);
+ }
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
+ return this.future.whenComplete((r, t) -> {
+ if (t != null) {
+ future.completeExceptionally(t);
+ } else {
+ future.complete(r);
+ }
+ });
+ }
+ }
+
+ /**
+ * <p>This is used to stage the unsent requests, i.e., {@link UnsentOffsetCommitRequest} and {@link UnsentOffsetFetchRequest}.
+ *
+ * <p>If the request is new. It will be enqueued and will be sent upon the next
+ * poll.
+ *
+ * <p>If the same request has been sent, the request's {@code CompletableFuture} will be completed upon the
+ * completion of the existing one.
+ *
+ * <p>There is a special handling for the {@link UnsentOffsetFetchRequest}. If a duplicated request was sent
+ * previously, we will chain the future to the current future.
+ */
+
+ class PendingRequests {
+ // Queue is used to ensure the sequence of commit
+ Queue<UnsentOffsetCommitRequest> unsentOffsetCommits = new LinkedList<>();
+ List<UnsentOffsetFetchRequest> unsentOffsetFetches = new ArrayList<>();
+ List<UnsentOffsetFetchRequest> inflightOffsetFetches = new ArrayList<>();
+
+ public boolean hasUnsentRequests() {
+ return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
+ }
+
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ UnsentOffsetCommitRequest request = new UnsentOffsetCommitRequest(
+ offsets,
+ groupState.groupId,
+ groupState.groupInstanceId.orElse(null),
+ groupState.generation);
+ unsentOffsetCommits.add(request);
+ return request.future();
+ }
+
+ /**
+ * We want to avoid duplication when sending an {@link OffsetFetchRequest}. The following checks are done:
+ * <li>1. dedupe against unsents: if a duplicated request was previously made, we chain the future</>
+ * <li>2. dedupe against incompleted: If a duplicated request was sent but hasn't gotten a results, we chain
+ * the future.</>
+ *
+ * <p>If the request is new, we chain a call back to remove itself from the {@code inflightOffsetFetches}
+ * upon completion.</>
+ */
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final UnsentOffsetFetchRequest request) {
+ Optional<UnsentOffsetFetchRequest> dupe =
+ unsentOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+ Optional<UnsentOffsetFetchRequest> inflight =
+ inflightOffsetFetches.stream().filter(r -> r.sameRequest(request)).findAny();
+
+ if (dupe.isPresent() || inflight.isPresent()) {
+ log.info("Duplicated OffsetFetchRequest: " + request.requestedPartitions);
+ dupe.orElseGet(() -> inflight.get()).chainFuture(request.future);
+ } else {
+ // remove the request from the outbound buffer: inflightOffsetFetches
+ request.future.whenComplete((r, t) -> {
+ if (!inflightOffsetFetches.remove(request)) {
+ log.info("unable to remove request from the outbound buffer:" + request);
+ }
+ });
+ this.unsentOffsetFetches.add(request);
+ }
+ return request.future;
+ }
+
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ UnsentOffsetFetchRequest request = new UnsentOffsetFetchRequest(
+ partitions,
+ groupState.generation,
+ retryBackoffMs);
+ return addOffsetFetchRequest(request);
+ }
+
+ public List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
+ List<NetworkClientDelegate.UnsentRequest> unsent = new ArrayList<>();
+ unsent.addAll(unsentOffsetCommits.stream()
+ .map(UnsentOffsetCommitRequest::toUnsentRequest)
+ .collect(Collectors.toList()));
+ List<UnsentOffsetFetchRequest> sendables = unsentOffsetFetches.stream()
+ .filter(r -> r.canSendRequest(currentTimeMs))
+ .collect(Collectors.toList());
+ inflightOffsetFetches.addAll(sendables);
+ unsent.addAll(sendables.stream()
+ .peek(r -> r.onSendAttempt(currentTimeMs))
+ .map(r -> r.toUnsentRequest(currentTimeMs))
+ .collect(Collectors.toList()));
+ // empty the staged requests
+ unsentOffsetCommits.clear();
Review Comment:
Ah yup! I was not watching the latest commits you had.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1133346162
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -437,29 +456,24 @@ public void wakeup() {
*/
@Override
public void commitSync(final Duration timeout) {
- final CommitApplicationEvent commitEvent = new CommitApplicationEvent(subscriptions.allConsumed());
- eventHandler.add(commitEvent);
-
- final CompletableFuture<Void> commitFuture = commitEvent.future();
- try {
- commitFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
- } catch (final TimeoutException e) {
- throw new org.apache.kafka.common.errors.TimeoutException(
- "timeout");
- } catch (final Exception e) {
- // handle exception here
- throw new RuntimeException(e);
- }
+ commitSync(subscriptions.allConsumed(), timeout);
}
@Override
public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets) {
- throw new KafkaException("method not implemented");
+ commitSync(offsets, Duration.ofMillis(defaultApiTimeoutMs));
}
@Override
public void commitSync(Map<TopicPartition, OffsetAndMetadata> offsets, Duration timeout) {
- throw new KafkaException("method not implemented");
+ CompletableFuture<Void> commitFuture = commit(offsets);
+ try {
+ commitFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (final TimeoutException e) {
Review Comment:
I'm a bit unsure about the exception handling 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1133442329
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -149,6 +185,29 @@ Queue<StagedCommit> stagedCommits() {
return this.stagedCommits;
}
+ /**
+ * Get all the sendable requests, and create a list of UnsentRequest.
+ */
+ private List<NetworkClientDelegate.UnsentRequest> sendOffsetFetchRequests(final long currentTimeMs) {
+ List<UnsentOffsetFetchRequestState> requests = unsentOffsetFetchRequests.sendableRequests(currentTimeMs);
+ List<NetworkClientDelegate.UnsentRequest> pollResults = new ArrayList<>();
Review Comment:
I could also do it in request.stream().map().reduce(). Let me know if one way is better than the other.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "guozhangwang (via GitHub)" <gi...@apache.org>.
guozhangwang commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1137549960
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
Review Comment:
Ah thanks for pointing it out, will check.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1139280595
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -235,4 +373,105 @@ public void ack(final long currentTimeMs) {
this.timer.update(currentTimeMs);
}
}
+
+ private class FetchCommittedOffsetResponseHandler {
Review Comment:
I think that's the original behavior of the code, I guess it does kind of make sense to retry on COORDINATOR_NOT_AVAILABLE, but I wonder if we did that purposely...
Here's the current code behavior:
https://github.com/apache/kafka/blob/d23ce20bdfbe5a9598523961cb7cf747ce4f52ef/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java#L1499
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #13380: KAFKA-14468: Committed API
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #13380:
URL: https://github.com/apache/kafka/pull/13380#discussion_r1139651358
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -137,31 +144,40 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
}
return null;
});
- return future;
}
- public void clientPoll(final long currentTimeMs) {
- this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
+ * {@link UnsentOffsetCommitRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<ClientResponse> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ return pendingRequests.addOffsetCommitRequest(offsets);
}
- // Visible for testing
- Queue<StagedCommit> stagedCommits() {
- return this.stagedCommits;
+ /**
+ * Handles {@link org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent}. It creates an
+ * {@link UnsentOffsetFetchRequest} and enqueue it to send later.
+ */
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ return pendingRequests.addOffsetFetchRequest(partitions);
+ }
+
+ public void clientPoll(final long currentTimeMs) {
+ this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs));
}
- private class StagedCommit {
+ private class UnsentOffsetCommitRequest {
Review Comment:
Rename to make naming consistent
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org