From 40abf2e5d5b5e0c7e45c8501339b64e42a7b5b7d Mon Sep 17 00:00:00 2001 From: Philip Nee Date: Mon, 5 Dec 2022 21:01:25 -0800 Subject: [PATCH 1/6] Add commit manager to handle commits commit manager clean up clean up clean up client poll to autocommit wip Addressed some PR comments Init commit for the request manager --- .../internals/CommitRequestManager.java | 243 ++++++++++++++++++ .../internals/DefaultBackgroundThread.java | 78 +++--- .../consumer/internals/GroupStateManager.java | 89 +++++++ .../internals/NetworkClientDelegate.java | 16 +- .../internals/PrototypeAsyncConsumer.java | 37 ++- .../consumer/internals/RequestManager.java | 4 + .../internals/events/ApplicationEvent.java | 2 +- .../events/ApplicationEventProcessor.java | 39 ++- .../events/CommitApplicationEvent.java | 63 +++++ .../events/PollApplicationEvent.java | 26 ++ .../internals/CommitRequestManagerTest.java | 101 ++++++++ .../DefaultBackgroundThreadTest.java | 38 ++- 12 files changed, 674 insertions(+), 62 deletions(-) create mode 100644 clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupStateManager.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/PollApplicationEvent.java create mode 100644 clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java new file mode 100644 index 0000000000000..a93da10bd7fd3 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java @@ -0,0 +1,243 @@ +/* + * 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; + +import org.apache.kafka.clients.ClientResponse; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.consumer.RetriableCommitFailedException; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.message.OffsetCommitRequestData; +import org.apache.kafka.common.record.RecordBatch; +import org.apache.kafka.common.requests.OffsetCommitRequest; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Timer; +import org.slf4j.Logger; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; + +public class CommitRequestManager implements RequestManager { + private final Queue stagedCommits; + // TODO: We will need to refactor the subscriptionState + private final SubscriptionState subscriptionState; + private final Logger log; + private final Optional autoCommitState; + private final CoordinatorRequestManager coordinatorRequestManager; + private final GroupStateManager groupState; + + public CommitRequestManager( + final Time time, + final LogContext logContext, + final SubscriptionState subscriptionState, + final ConsumerConfig config, + final CoordinatorRequestManager coordinatorRequestManager, + final GroupStateManager groupState) { + this.log = logContext.logger(getClass()); + this.stagedCommits = new LinkedList<>(); + if (config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) { + final long autoCommitInterval = + Integer.toUnsignedLong(config.getInt(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG)); + this.autoCommitState = Optional.of(new AutoCommitState(time, autoCommitInterval)); + } else { + this.autoCommitState = Optional.empty(); + } + this.coordinatorRequestManager = coordinatorRequestManager; + this.groupState = groupState; + this.subscriptionState = subscriptionState; + } + + // Visible for testing + CommitRequestManager( + final LogContext logContext, + final SubscriptionState subscriptionState, + final CoordinatorRequestManager coordinatorRequestManager, + final GroupStateManager groupState, + final AutoCommitState autoCommitState) { + this.log = logContext.logger(getClass()); + this.subscriptionState = subscriptionState; + this.coordinatorRequestManager = coordinatorRequestManager; + this.groupState = groupState; + this.autoCommitState = Optional.ofNullable(autoCommitState); + this.stagedCommits = new LinkedList<>(); + } + + /** + * Poll for the commit request if there's any. The function will also try to autocommit, if enabled. + * + * @param currentTimeMs + * @return + */ + @Override + public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { + if (coordinatorRequestManager == null) { + return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>()); + } + + maybeAutoCommit(currentTimeMs); + + if (stagedCommits.isEmpty()) { + return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>()); + } + + List unsentCommitRequests = + stagedCommits.stream().map(StagedCommit::toUnsentRequest).collect(Collectors.toList()); + return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.unmodifiableList(unsentCommitRequests)); + } + + public CompletableFuture add(final Map offsets) { + StagedCommit commit = new StagedCommit( + offsets, + groupState.groupId, + groupState.groupInstanceId.orElse(null), + groupState.generation); + this.stagedCommits.add(commit); + return commit.future(); + } + + private void maybeAutoCommit(final long currentTimeMs) { + if (!autoCommitState.isPresent()) { + return; + } + + AutoCommitState autocommit = autoCommitState.get(); + if (!autocommit.canSendAutocommit(currentTimeMs)) { + return; + } + + Map allConsumedOffsets = subscriptionState.allConsumed(); + log.debug("Auto-committing offsets {}", allConsumedOffsets); + this.add(allConsumedOffsets).whenComplete((response, throwable) -> { + if (throwable == null) { + log.debug("Completed asynchronous auto-commit of offsets {}", allConsumedOffsets); + } + + if (throwable instanceof RetriableCommitFailedException) { + log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", allConsumedOffsets, + throwable); + autoCommitState.get().reset(); + } else { + log.warn("Asynchronous auto-commit of offsets {} failed: {}", allConsumedOffsets, + throwable.getMessage()); + } + }); + autocommit.reset(); + } + + public void clientPoll(final long currentTimeMs) { + this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs)); + } + + private static class OffsetCommitRequestHandler extends NetworkClientDelegate.FutureCompletionHandler { + protected final Map offsets; + + public OffsetCommitRequestHandler(final Map offsets) { + this.offsets = offsets; + } + } + + private class StagedCommit { + private final Map offsets; + private final String groupId; + private final GroupStateManager.Generation generation; + private final String groupInstanceId; + private final NetworkClientDelegate.FutureCompletionHandler callback; + + public StagedCommit(final Map offsets, + final String groupId, + final String groupInstanceId, + final GroupStateManager.Generation generation) { + this.offsets = offsets; + // if no callback is provided, DefaultOffsetCommitCallback will be used. + this.callback = new NetworkClientDelegate.FutureCompletionHandler(); + this.groupId = groupId; + this.generation = generation; + this.groupInstanceId = groupInstanceId; + } + + public CompletableFuture future() { + return callback.future(); + } + + public NetworkClientDelegate.UnsentRequest toUnsentRequest() { + Map requestTopicDataMap = new HashMap<>(); + for (Map.Entry entry : offsets.entrySet()) { + TopicPartition topicPartition = entry.getKey(); + OffsetAndMetadata offsetAndMetadata = entry.getValue(); + + OffsetCommitRequestData.OffsetCommitRequestTopic topic = requestTopicDataMap + .getOrDefault(topicPartition.topic(), + new OffsetCommitRequestData.OffsetCommitRequestTopic() + .setName(topicPartition.topic()) + ); + + topic.partitions().add(new OffsetCommitRequestData.OffsetCommitRequestPartition() + .setPartitionIndex(topicPartition.partition()) + .setCommittedOffset(offsetAndMetadata.offset()) + .setCommittedLeaderEpoch(offsetAndMetadata.leaderEpoch().orElse(RecordBatch.NO_PARTITION_LEADER_EPOCH)) + .setCommittedMetadata(offsetAndMetadata.metadata()) + ); + requestTopicDataMap.put(topicPartition.topic(), topic); + } + + OffsetCommitRequest.Builder builder = new OffsetCommitRequest.Builder( + new OffsetCommitRequestData() + .setGroupId(this.groupId) + .setGenerationId(generation.generationId) + .setMemberId(generation.memberId) + .setGroupInstanceId(groupInstanceId) + .setTopics(new ArrayList<>(requestTopicDataMap.values()))); + return new NetworkClientDelegate.UnsentRequest( + builder, + coordinatorRequestManager.coordinator(), + callback); + } + } + + static class AutoCommitState { + private final Timer timer; + private final long autoCommitInterval; + + public AutoCommitState( + final Time time, + final long autoCommitInterval) { + this.autoCommitInterval = autoCommitInterval; + this.timer = time.timer(autoCommitInterval); + } + + public boolean canSendAutocommit(final long currentTimeMs) { + return this.timer.isExpired(); + } + + public void reset() { + this.timer.reset(autoCommitInterval); + } + + public void ack(final long currentTimeMs) { + this.timer.update(currentTimeMs); + } + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java index 481137e074786..7790aa89b5ae1 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java @@ -30,8 +30,10 @@ import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; -import java.util.Iterator; +import java.util.Collections; +import java.util.HashMap; import java.util.LinkedList; +import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.Queue; @@ -46,7 +48,7 @@ * initialized by the polling thread. */ public class DefaultBackgroundThread extends KafkaThread { - private static final int MAX_POLL_TIMEOUT_MS = 5000; + private static final long MAX_POLL_TIMEOUT_MS = 5000; private static final String BACKGROUND_THREAD_NAME = "consumer_background_thread"; private final Time time; @@ -56,13 +58,14 @@ public class DefaultBackgroundThread extends KafkaThread { private final ConsumerMetadata metadata; private final ConsumerConfig config; // empty if groupId is null - private final Optional coordinatorManager; private final ApplicationEventProcessor applicationEventProcessor; private final NetworkClientDelegate networkClientDelegate; private final ErrorEventHandler errorEventHandler; - + private final GroupStateManager groupState; private boolean running; + private final Map> requestManagerRegistry; + // Visible for testing DefaultBackgroundThread(final Time time, final ConsumerConfig config, @@ -73,7 +76,9 @@ public class DefaultBackgroundThread extends KafkaThread { final ApplicationEventProcessor processor, final ConsumerMetadata metadata, final NetworkClientDelegate networkClient, - final CoordinatorRequestManager coordinatorManager) { + final GroupStateManager groupState, + final CoordinatorRequestManager coordinatorManager, + CommitRequestManager commitRequestManager) { super(BACKGROUND_THREAD_NAME, true); this.time = time; this.running = true; @@ -84,8 +89,12 @@ public class DefaultBackgroundThread extends KafkaThread { this.config = config; this.metadata = metadata; this.networkClientDelegate = networkClient; - this.coordinatorManager = Optional.ofNullable(coordinatorManager); this.errorEventHandler = errorEventHandler; + this.groupState = groupState; + + this.requestManagerRegistry = new HashMap<>(); + this.requestManagerRegistry.put(RequestManager.Type.COORDINATOR, Optional.ofNullable(coordinatorManager)); + this.requestManagerRegistry.put(RequestManager.Type.COMMIT, Optional.ofNullable(commitRequestManager)); } public DefaultBackgroundThread(final Time time, final ConsumerConfig config, @@ -111,23 +120,37 @@ public DefaultBackgroundThread(final Time time, networkClient); this.running = true; this.errorEventHandler = new ErrorEventHandler(this.backgroundEventQueue); - String groupId = rebalanceConfig.groupId; - this.coordinatorManager = groupId == null ? - Optional.empty() : - Optional.of(new CoordinatorRequestManager( - time, - logContext, - config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG), - errorEventHandler, - groupId - )); - this.applicationEventProcessor = new ApplicationEventProcessor(backgroundEventQueue); + this.groupState = new GroupStateManager(rebalanceConfig); + this.requestManagerRegistry = Collections.unmodifiableMap(buildRequestManagerRegistry(logContext)); + this.applicationEventProcessor = new ApplicationEventProcessor(backgroundEventQueue, requestManagerRegistry); } catch (final Exception e) { close(); throw new KafkaException("Failed to construct background processor", e.getCause()); } } + private Map> buildRequestManagerRegistry(final LogContext logContext) { + Map> registry = new HashMap<>(); + CoordinatorRequestManager coordinatorManager = groupState.groupId == null ? + null : + new CoordinatorRequestManager( + time, + logContext, + config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG), + errorEventHandler, + groupState.groupId); + // Add subscriptionState + CommitRequestManager commitRequestManager = coordinatorManager == null ? + null : + new CommitRequestManager(time, + logContext, null, config, + coordinatorManager, + groupState); + registry.put(RequestManager.Type.COORDINATOR, Optional.ofNullable(coordinatorManager)); + registry.put(RequestManager.Type.COMMIT, Optional.ofNullable(commitRequestManager)); + return registry; + } + @Override public void run() { try { @@ -151,30 +174,23 @@ public void run() { /** * Poll and process an {@link ApplicationEvent}. It performs the following tasks: - * 1. Drains and try to process all of the requests in the queue. - * 2. Poll request managers to queue up the necessary requests. + * 1. Drains and try to process all the requests in the queue. + * 2. Iterate through the registry, poll, and get the next poll time for the network poll * 3. Poll the networkClient to send and retrieve the response. */ void runOnce() { drainAndProcess(); - final long currentTimeMs = time.milliseconds(); - long pollWaitTimeMs = MAX_POLL_TIMEOUT_MS; - - if (coordinatorManager.isPresent()) { - pollWaitTimeMs = Math.min( - pollWaitTimeMs, - handlePollResult(coordinatorManager.get().poll(currentTimeMs))); - } - + final long pollWaitTimeMs = requestManagerRegistry.values().stream() + .filter(Optional::isPresent) + .map(m -> handlePollResult(m.get().poll(currentTimeMs))) + .reduce(MAX_POLL_TIMEOUT_MS, Math::min); networkClientDelegate.poll(pollWaitTimeMs, currentTimeMs); } private void drainAndProcess() { Queue events = pollApplicationEvent(); - Iterator iter = events.iterator(); - while (iter.hasNext()) { - ApplicationEvent event = iter.next(); + for (ApplicationEvent event : events) { log.debug("processing application event: {}", event); consumeApplicationEvent(event); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupStateManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupStateManager.java new file mode 100644 index 0000000000000..f7c0083c5137d --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupStateManager.java @@ -0,0 +1,89 @@ +/* + * 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; + +import org.apache.kafka.clients.GroupRebalanceConfig; +import org.apache.kafka.common.requests.JoinGroupRequest; +import org.apache.kafka.common.requests.OffsetCommitRequest; + +import java.util.Objects; +import java.util.Optional; + +public class GroupStateManager { + public final String groupId; + public final Optional groupInstanceId; + public Generation generation = Generation.NO_GENERATION; + + public GroupStateManager(String groupId, Optional groupInstanceId) { + this.groupId = groupId; + this.groupInstanceId = groupInstanceId; + } + + public GroupStateManager(final GroupRebalanceConfig config) { + this.groupId = config.groupId; + this.groupInstanceId = config.groupInstanceId; + } + + protected static class Generation { + public static final Generation NO_GENERATION = new Generation( + OffsetCommitRequest.DEFAULT_GENERATION_ID, + JoinGroupRequest.UNKNOWN_MEMBER_ID, + null); + + public final int generationId; + public final String memberId; + public final String protocolName; + + public Generation(int generationId, String memberId, String protocolName) { + this.generationId = generationId; + this.memberId = memberId; + this.protocolName = protocolName; + } + + /** + * @return true if this generation has a valid member id, false otherwise. A member might have an id before + * it becomes part of a group generation. + */ + public boolean hasMemberId() { + return !memberId.isEmpty(); + } + + @Override + public boolean equals(final Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + final Generation that = (Generation) o; + return generationId == that.generationId && + Objects.equals(memberId, that.memberId) && + Objects.equals(protocolName, that.protocolName); + } + + @Override + public int hashCode() { + return Objects.hash(generationId, memberId, protocolName); + } + + @Override + public String toString() { + return "Generation{" + + "generationId=" + generationId + + ", memberId='" + memberId + '\'' + + ", protocol='" + protocolName + '\'' + + '}'; + } + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java index fd36861a2edc3..c817b942e2e86 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java @@ -201,9 +201,15 @@ public static class UnsentRequest { private Timer timer; public UnsentRequest( - final AbstractRequest.Builder requestBuilder, - final Optional node - ) { + final AbstractRequest.Builder requestBuilder, + final Optional node) { + this(requestBuilder, node, new FutureCompletionHandler()); + } + + public UnsentRequest( + final AbstractRequest.Builder requestBuilder, + final Optional node, + final FutureCompletionHandler handler) { Objects.requireNonNull(requestBuilder); this.requestBuilder = requestBuilder; this.node = node; @@ -243,6 +249,10 @@ public void onFailure(final RuntimeException e) { future.completeExceptionally(e); } + public CompletableFuture future() { + return future; + } + @Override public void onComplete(final ClientResponse response) { if (response.authenticationException() != null) { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java index 9c66c8a76eae6..18fd79621a946 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java @@ -29,8 +29,8 @@ import org.apache.kafka.clients.consumer.OffsetAndTimestamp; import org.apache.kafka.clients.consumer.OffsetCommitCallback; import org.apache.kafka.clients.consumer.OffsetResetStrategy; -import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent; import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent; +import org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent; import org.apache.kafka.clients.consumer.internals.events.EventHandler; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.Metric; @@ -216,18 +216,21 @@ private Fetch collectFetches() { */ @Override public void commitAsync() { - final ApplicationEvent commitEvent = new CommitApplicationEvent(); - eventHandler.add(commitEvent); + commitAsync(null); } @Override public void commitAsync(OffsetCommitCallback callback) { - throw new KafkaException("method not implemented"); + commitAsync(subscriptions.allConsumed(), callback); } @Override public void commitAsync(Map offsets, OffsetCommitCallback callback) { - throw new KafkaException("method not implemented"); + final CommitApplicationEvent commitEvent = new CommitApplicationEvent(offsets); + commitEvent.future().whenComplete((r, t) -> { + callback.onComplete(offsets, new RuntimeException(t)); + }); + eventHandler.add(commitEvent); } @Override @@ -395,10 +398,10 @@ public void wakeup() { */ @Override public void commitSync(final Duration timeout) { - final CommitApplicationEvent commitEvent = new CommitApplicationEvent(); + final CommitApplicationEvent commitEvent = new CommitApplicationEvent(subscriptions.allConsumed()); eventHandler.add(commitEvent); - final CompletableFuture commitFuture = commitEvent.commitFuture; + final CompletableFuture commitFuture = commitEvent.future(); try { commitFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS); } catch (final TimeoutException e) { @@ -471,18 +474,6 @@ public ConsumerRecords poll(long timeout) { throw new KafkaException("method not implemented"); } - /** - * A stubbed ApplicationEvent for demonstration purpose - */ - private class CommitApplicationEvent extends ApplicationEvent { - // this is the stubbed commitAsyncEvents - CompletableFuture commitFuture = new CompletableFuture<>(); - - public CommitApplicationEvent() { - super(Type.COMMIT); - } - } - private static ClusterResourceListeners configureClusterResourceListeners( final Deserializer keyDeserializer, final Deserializer valueDeserializer, @@ -512,4 +503,12 @@ private static Metrics buildMetrics( config.originalsWithPrefix(CommonClientConfigs.METRICS_CONTEXT_PREFIX)); return new Metrics(metricConfig, reporters, time, metricsContext); } + + private class DefaultOffsetCommitCallback implements OffsetCommitCallback { + @Override + public void onComplete(Map offsets, Exception exception) { + if (exception != null) + log.error("Offset commit with offsets {} failed", offsets, exception); + } + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManager.java index 13c7c4d9d16b4..a9c381fadc7c9 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManager.java @@ -24,4 +24,8 @@ */ public interface RequestManager { PollResult poll(long currentTimeMs); + + enum Type { + COORDINATOR, COMMIT + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java index 2218441485c38..d62278b9b3fbb 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java @@ -36,6 +36,6 @@ public String toString() { return type + " ApplicationEvent"; } public enum Type { - NOOP, COMMIT, + NOOP, COMMIT, POLL, } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java index e031fb9f7bbce..6f903c64dfac2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java @@ -16,22 +16,37 @@ */ package org.apache.kafka.clients.consumer.internals.events; +import org.apache.kafka.clients.ClientResponse; +import org.apache.kafka.clients.consumer.internals.CommitRequestManager; import org.apache.kafka.clients.consumer.internals.NoopBackgroundEvent; +import org.apache.kafka.clients.consumer.internals.RequestManager; +import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; public class ApplicationEventProcessor { private final BlockingQueue backgroundEventQueue; + private final Map> registry; - public ApplicationEventProcessor(final BlockingQueue backgroundEventQueue) { + public ApplicationEventProcessor( + final BlockingQueue backgroundEventQueue, + final Map> requestManagerRegistry) { this.backgroundEventQueue = backgroundEventQueue; + this.registry = requestManagerRegistry; } + public boolean process(final ApplicationEvent event) { Objects.requireNonNull(event); switch (event.type) { case NOOP: return process((NoopApplicationEvent) event); + case COMMIT: + return process((CommitApplicationEvent) event); + case POLL: + return process((PollApplicationEvent) event); } return false; } @@ -46,4 +61,26 @@ public boolean process(final ApplicationEvent event) { private boolean process(final NoopApplicationEvent event) { return backgroundEventQueue.add(new NoopBackgroundEvent(event.message)); } + + private boolean process(final PollApplicationEvent event) { + Optional commitRequestManger = registry.get(RequestManager.Type.COMMIT); + if (!commitRequestManger.isPresent()) { + return false; + } + + CommitRequestManager manager = (CommitRequestManager) commitRequestManger.get(); + manager.clientPoll(event.pollTimeMs); + return true; + } + + private boolean process(final CommitApplicationEvent event) { + Optional commitRequestManger = registry.get(RequestManager.Type.COMMIT); + if (!commitRequestManger.isPresent()) { + return false; + } + + CommitRequestManager manager = (CommitRequestManager) commitRequestManger.get(); + CompletableFuture res = manager.add(event.offsets()); + return true; + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java new file mode 100644 index 0000000000000..562126e972b0d --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java @@ -0,0 +1,63 @@ +/* + * 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.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; + +public class CommitApplicationEvent extends ApplicationEvent { + final private CompletableFuture future; + final private Map offsets; + + public CommitApplicationEvent(final Map offsets) { + super(Type.COMMIT); + this.offsets = offsets; + Optional exception = isValid(offsets); + if (exception.isPresent()) { + throw new RuntimeException(exception.get()); + } + this.future = new CompletableFuture<>(); + } + + public CompletableFuture future() { + return future; + } + + public Map offsets() { + return offsets; + } + + private Optional isValid(final Map offsets) { + for (Map.Entry entry : offsets.entrySet()) { + TopicPartition topicPartition = entry.getKey(); + OffsetAndMetadata offsetAndMetadata = entry.getValue(); + if (offsetAndMetadata.offset() < 0) { + return Optional.of(new IllegalArgumentException("Invalid offset: " + offsetAndMetadata.offset())); + } + } + return Optional.empty(); + } + + @Override + public String toString() { + return getClass() + "_" + this.offsets; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/PollApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/PollApplicationEvent.java new file mode 100644 index 0000000000000..c14998dc65243 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/PollApplicationEvent.java @@ -0,0 +1,26 @@ +/* + * 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; + +public class PollApplicationEvent extends ApplicationEvent { + public final long pollTimeMs; + + protected PollApplicationEvent(final long currentTimeMs) { + super(Type.POLL); + this.pollTimeMs = currentTimeMs; + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java new file mode 100644 index 0000000000000..f5784757f7154 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java @@ -0,0 +1,101 @@ +/* + * 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; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.MockTime; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.Properties; + +import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class CommitRequestManagerTest { + private SubscriptionState subscriptionState; + private GroupStateManager groupStateManager; + private LogContext logContext; + private MockTime time; + private CoordinatorRequestManager coordinatorRequestManager; + private Properties props; + + @BeforeEach + public void setup() { + this.logContext = new LogContext(); + this.time = new MockTime(0); + this.subscriptionState = mock(SubscriptionState.class); + this.coordinatorRequestManager = mock(CoordinatorRequestManager.class); + this.groupStateManager = new GroupStateManager("group-1", Optional.empty()); + + this.props = new Properties(); + this.props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); + this.props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + this.props.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + } + + @Test + public void testPoll() { + CommitRequestManager commitRequestManger = create(false, 0); + NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds()); + assertEquals(0, res.unsentRequests.size()); + assertEquals(Long.MAX_VALUE, res.timeUntilNextPollMs); + + Map 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()); + } + + @Test + public void testPollAndAutocommit() { + CommitRequestManager commitRequestManger = create(true, 100); + NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds()); + assertEquals(0, res.unsentRequests.size()); + assertEquals(Long.MAX_VALUE, res.timeUntilNextPollMs); + + Map offsets = new HashMap<>(); + offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0)); + commitRequestManger.clientPoll(time.milliseconds()); + when(subscriptionState.allConsumed()).thenReturn(offsets); + time.sleep(100); + commitRequestManger.clientPoll(time.milliseconds()); + res = commitRequestManger.poll(time.milliseconds()); + assertEquals(1, res.unsentRequests.size()); + } + + private CommitRequestManager create(final boolean autoCommitEnabled, final long autoCommitInterval) { + return new CommitRequestManager( + this.time, + this.logContext, + this.subscriptionState, + new ConsumerConfig(props), + this.coordinatorRequestManager, + this.groupStateManager); + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java index 7ee4ee6a1263d..71970c0650d98 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.clients.consumer.internals; +import org.apache.kafka.clients.GroupRebalanceConfig; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent; import org.apache.kafka.clients.consumer.internals.events.ApplicationEventProcessor; @@ -50,7 +51,7 @@ import static org.mockito.Mockito.when; public class DefaultBackgroundThreadTest { - private static final long REFRESH_BACK_OFF_MS = 100; + private static final long RETRY_BACKOFF_MS = 100; private final Properties properties = new Properties(); private MockTime time; private ConsumerMetadata metadata; @@ -61,6 +62,8 @@ public class DefaultBackgroundThreadTest { private CoordinatorRequestManager coordinatorManager; private ErrorEventHandler errorEventHandler; private int requestTimeoutMs = 500; + private GroupStateManager groupState; + private CommitRequestManager commitManager; @BeforeEach @SuppressWarnings("unchecked") @@ -73,6 +76,16 @@ public void setup() { this.processor = mock(ApplicationEventProcessor.class); this.coordinatorManager = mock(CoordinatorRequestManager.class); this.errorEventHandler = mock(ErrorEventHandler.class); + GroupRebalanceConfig rebalanceConfig = new GroupRebalanceConfig( + 100, + 100, + 100, + "group_id", + Optional.empty(), + 100, + true); + this.groupState = new GroupStateManager(rebalanceConfig); + this.commitManager = mock(CommitRequestManager.class); } @Test @@ -87,7 +100,8 @@ public void testStartupAndTearDown() { public void testApplicationEvent() { this.applicationEventsQueue = new LinkedBlockingQueue<>(); this.backgroundEventsQueue = new LinkedBlockingQueue<>(); - when(coordinatorManager.poll(anyLong())).thenReturn(mockPollResult()); + when(coordinatorManager.poll(anyLong())).thenReturn(mockPollCoordinatorResult()); + when(commitManager.poll(anyLong())).thenReturn(mockPollCommitResult()); DefaultBackgroundThread backgroundThread = mockBackgroundThread(); ApplicationEvent e = new NoopApplicationEvent("noop event"); this.applicationEventsQueue.add(e); @@ -99,9 +113,11 @@ public void testApplicationEvent() { @Test void testFindCoordinator() { DefaultBackgroundThread backgroundThread = mockBackgroundThread(); - when(this.coordinatorManager.poll(time.milliseconds())).thenReturn(mockPollResult()); + when(this.coordinatorManager.poll(time.milliseconds())).thenReturn(mockPollCoordinatorResult()); + when(this.commitManager.poll(time.milliseconds())).thenReturn(mockPollCommitResult()); backgroundThread.runOnce(); Mockito.verify(coordinatorManager, times(1)).poll(anyLong()); + Mockito.verify(commitManager, times(1)).poll(anyLong()); Mockito.verify(networkClient, times(1)).poll(anyLong(), anyLong()); backgroundThread.close(); } @@ -138,7 +154,7 @@ private static NetworkClientDelegate.UnsentRequest findCoordinatorUnsentRequest( private DefaultBackgroundThread mockBackgroundThread() { properties.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); properties.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); - properties.put(RETRY_BACKOFF_MS_CONFIG, REFRESH_BACK_OFF_MS); + properties.put(RETRY_BACKOFF_MS_CONFIG, RETRY_BACKOFF_MS); return new DefaultBackgroundThread( this.time, @@ -150,12 +166,20 @@ private DefaultBackgroundThread mockBackgroundThread() { processor, this.metadata, this.networkClient, - this.coordinatorManager); + this.groupState, + this.coordinatorManager, + this.commitManager); } - private NetworkClientDelegate.PollResult mockPollResult() { + private NetworkClientDelegate.PollResult mockPollCoordinatorResult() { return new NetworkClientDelegate.PollResult( - 0, + RETRY_BACKOFF_MS, + Collections.singletonList(findCoordinatorUnsentRequest(time, requestTimeoutMs))); + } + + private NetworkClientDelegate.PollResult mockPollCommitResult() { + return new NetworkClientDelegate.PollResult( + RETRY_BACKOFF_MS, Collections.singletonList(findCoordinatorUnsentRequest(time, requestTimeoutMs))); } } From a3d87af3db2203719d503fbe5c206d7fa61dfa7d Mon Sep 17 00:00:00 2001 From: Philip Nee Date: Tue, 24 Jan 2023 16:05:12 -0800 Subject: [PATCH 2/6] dead store --- .../consumer/internals/events/ApplicationEventProcessor.java | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java index 6f903c64dfac2..5e015bdcfc863 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java @@ -16,7 +16,6 @@ */ package org.apache.kafka.clients.consumer.internals.events; -import org.apache.kafka.clients.ClientResponse; import org.apache.kafka.clients.consumer.internals.CommitRequestManager; import org.apache.kafka.clients.consumer.internals.NoopBackgroundEvent; import org.apache.kafka.clients.consumer.internals.RequestManager; @@ -25,7 +24,6 @@ import java.util.Objects; import java.util.Optional; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CompletableFuture; public class ApplicationEventProcessor { private final BlockingQueue backgroundEventQueue; @@ -80,7 +78,7 @@ private boolean process(final CommitApplicationEvent event) { } CommitRequestManager manager = (CommitRequestManager) commitRequestManger.get(); - CompletableFuture res = manager.add(event.offsets()); + manager.add(event.offsets()); return true; } } From 49aa05d343f672d940a6efd768101452070ee436 Mon Sep 17 00:00:00 2001 From: Philip Nee Date: Fri, 27 Jan 2023 12:51:34 -0800 Subject: [PATCH 3/6] Addressed previous comments and add more test cases to improve coverage --- .../internals/CommitRequestManager.java | 49 ++++++------------- .../internals/DefaultBackgroundThread.java | 10 ++-- .../events/ApplicationEventProcessor.java | 12 ++++- .../internals/CommitRequestManagerTest.java | 10 +++- .../DefaultBackgroundThreadTest.java | 9 ++-- 5 files changed, 44 insertions(+), 46 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java index a93da10bd7fd3..4de8ac960e4f2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java @@ -35,6 +35,7 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Queue; import java.util.concurrent.CompletableFuture; @@ -56,6 +57,7 @@ public CommitRequestManager( final ConsumerConfig config, final CoordinatorRequestManager coordinatorRequestManager, final GroupStateManager groupState) { + Objects.requireNonNull(coordinatorRequestManager, "Coordinator is needed upon committing offsets"); this.log = logContext.logger(getClass()); this.stagedCommits = new LinkedList<>(); if (config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) { @@ -70,21 +72,6 @@ public CommitRequestManager( this.subscriptionState = subscriptionState; } - // Visible for testing - CommitRequestManager( - final LogContext logContext, - final SubscriptionState subscriptionState, - final CoordinatorRequestManager coordinatorRequestManager, - final GroupStateManager groupState, - final AutoCommitState autoCommitState) { - this.log = logContext.logger(getClass()); - this.subscriptionState = subscriptionState; - this.coordinatorRequestManager = coordinatorRequestManager; - this.groupState = groupState; - this.autoCommitState = Optional.ofNullable(autoCommitState); - this.stagedCommits = new LinkedList<>(); - } - /** * Poll for the commit request if there's any. The function will also try to autocommit, if enabled. * @@ -93,10 +80,6 @@ public CommitRequestManager( */ @Override public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { - if (coordinatorRequestManager == null) { - return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>()); - } - maybeAutoCommit(currentTimeMs); if (stagedCommits.isEmpty()) { @@ -130,7 +113,14 @@ private void maybeAutoCommit(final long currentTimeMs) { Map allConsumedOffsets = subscriptionState.allConsumed(); log.debug("Auto-committing offsets {}", allConsumedOffsets); - this.add(allConsumedOffsets).whenComplete((response, throwable) -> { + sendAutoCommit(allConsumedOffsets); + autocommit.reset(); + } + + // Visible for testing + CompletableFuture sendAutoCommit(final Map allConsumedOffsets) { + CompletableFuture future = this.add(allConsumedOffsets); + future.whenComplete((response, throwable) -> { if (throwable == null) { log.debug("Completed asynchronous auto-commit of offsets {}", allConsumedOffsets); } @@ -138,33 +128,24 @@ private void maybeAutoCommit(final long currentTimeMs) { if (throwable instanceof RetriableCommitFailedException) { log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", allConsumedOffsets, throwable); - autoCommitState.get().reset(); } else { log.warn("Asynchronous auto-commit of offsets {} failed: {}", allConsumedOffsets, throwable.getMessage()); } }); - autocommit.reset(); + return future; } public void clientPoll(final long currentTimeMs) { this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs)); } - private static class OffsetCommitRequestHandler extends NetworkClientDelegate.FutureCompletionHandler { - protected final Map offsets; - - public OffsetCommitRequestHandler(final Map offsets) { - this.offsets = offsets; - } - } - private class StagedCommit { private final Map offsets; private final String groupId; private final GroupStateManager.Generation generation; private final String groupInstanceId; - private final NetworkClientDelegate.FutureCompletionHandler callback; + private final NetworkClientDelegate.FutureCompletionHandler future; public StagedCommit(final Map offsets, final String groupId, @@ -172,14 +153,14 @@ public StagedCommit(final Map offsets, final GroupStateManager.Generation generation) { this.offsets = offsets; // if no callback is provided, DefaultOffsetCommitCallback will be used. - this.callback = new NetworkClientDelegate.FutureCompletionHandler(); + this.future = new NetworkClientDelegate.FutureCompletionHandler(); this.groupId = groupId; this.generation = generation; this.groupInstanceId = groupInstanceId; } public CompletableFuture future() { - return callback.future(); + return future.future(); } public NetworkClientDelegate.UnsentRequest toUnsentRequest() { @@ -213,7 +194,7 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() { return new NetworkClientDelegate.UnsentRequest( builder, coordinatorRequestManager.coordinator(), - callback); + future); } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java index 7790aa89b5ae1..b565b33e676b0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java @@ -179,19 +179,20 @@ public void run() { * 3. Poll the networkClient to send and retrieve the response. */ void runOnce() { - drainAndProcess(); + drain(); final long currentTimeMs = time.milliseconds(); final long pollWaitTimeMs = requestManagerRegistry.values().stream() .filter(Optional::isPresent) - .map(m -> handlePollResult(m.get().poll(currentTimeMs))) + .map(m -> m.get().poll(currentTimeMs)) + .map(this::handlePollResult) .reduce(MAX_POLL_TIMEOUT_MS, Math::min); networkClientDelegate.poll(pollWaitTimeMs, currentTimeMs); } - private void drainAndProcess() { + private void drain() { Queue events = pollApplicationEvent(); for (ApplicationEvent event : events) { - log.debug("processing application event: {}", event); + log.debug("Consuming application event: {}", event); consumeApplicationEvent(event); } } @@ -214,7 +215,6 @@ private Queue pollApplicationEvent() { } private void consumeApplicationEvent(final ApplicationEvent event) { - log.debug("try consuming event: {}", Optional.ofNullable(event)); Objects.requireNonNull(event); applicationEventProcessor.process(event); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java index 5e015bdcfc863..0d005170d5f70 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java @@ -74,11 +74,21 @@ private boolean process(final PollApplicationEvent event) { private boolean process(final CommitApplicationEvent event) { Optional commitRequestManger = registry.get(RequestManager.Type.COMMIT); if (!commitRequestManger.isPresent()) { + // Leaving this error handling here, but it is a bit strange as the commit API should enforce the group.id + // upfront so we should never get to this block. + backgroundEventQueue.add(new ErrorBackgroundEvent( + new RuntimeException("Unable to commit offset. Most likely because the group.id wasn't set"))); return false; } CommitRequestManager manager = (CommitRequestManager) commitRequestManger.get(); - manager.add(event.offsets()); + manager.add(event.offsets()).whenComplete((r, e) -> { + if (e != null) { + event.future().completeExceptionally(e); + return; + } + event.future().complete(null); + }); return true; } } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java index f5784757f7154..6694919463abc 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java @@ -73,7 +73,7 @@ public void testPoll() { } @Test - public void testPollAndAutocommit() { + public void testPollAndAutoCommit() { CommitRequestManager commitRequestManger = create(true, 100); NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds()); assertEquals(0, res.unsentRequests.size()); @@ -89,6 +89,14 @@ public void testPollAndAutocommit() { assertEquals(1, res.unsentRequests.size()); } + @Test + public void testAutoCommitFuture() { + CommitRequestManager commitRequestManger = create(true, 100); + commitRequestManger.sendAutoCommit(new HashMap<>()).complete(null); + commitRequestManger.sendAutoCommit(new HashMap<>()).completeExceptionally(new RuntimeException("mock " + + "exception")); + } + private CommitRequestManager create(final boolean autoCommitEnabled, final long autoCommitInterval) { return new CommitRequestManager( this.time, diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java index 71970c0650d98..73d29b364b3aa 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java @@ -113,11 +113,10 @@ public void testApplicationEvent() { @Test void testFindCoordinator() { DefaultBackgroundThread backgroundThread = mockBackgroundThread(); - when(this.coordinatorManager.poll(time.milliseconds())).thenReturn(mockPollCoordinatorResult()); - when(this.commitManager.poll(time.milliseconds())).thenReturn(mockPollCommitResult()); + when(this.coordinatorManager.poll(anyLong())).thenReturn(mockPollCoordinatorResult()); + when(this.commitManager.poll(anyLong())).thenReturn(mockPollCommitResult()); backgroundThread.runOnce(); Mockito.verify(coordinatorManager, times(1)).poll(anyLong()); - Mockito.verify(commitManager, times(1)).poll(anyLong()); Mockito.verify(networkClient, times(1)).poll(anyLong(), anyLong()); backgroundThread.close(); } @@ -138,8 +137,8 @@ void testPollResultTimer() { } private static NetworkClientDelegate.UnsentRequest findCoordinatorUnsentRequest( - final Time time, - final long timeout + final Time time, + final long timeout ) { NetworkClientDelegate.UnsentRequest req = new NetworkClientDelegate.UnsentRequest( new FindCoordinatorRequest.Builder( From 9813dcf5055f6d0f082127ef8dd24073e9841611 Mon Sep 17 00:00:00 2001 From: Philip Nee Date: Mon, 6 Feb 2023 13:10:01 -0800 Subject: [PATCH 4/6] Addressed some comments Address comments from the PR review --- .../internals/CommitRequestManager.java | 55 +++++++++++-------- .../internals/DefaultBackgroundThread.java | 6 +- ...GroupStateManager.java => GroupState.java} | 6 +- .../events/ApplicationEventProcessor.java | 2 +- .../events/CommitApplicationEvent.java | 3 +- .../internals/CommitRequestManagerTest.java | 26 ++++++++- .../DefaultBackgroundThreadTest.java | 4 +- 7 files changed, 66 insertions(+), 36 deletions(-) rename clients/src/main/java/org/apache/kafka/clients/consumer/internals/{GroupStateManager.java => GroupState.java} (94%) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java index 4de8ac960e4f2..cb05959bdbc2c 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java @@ -48,7 +48,7 @@ public class CommitRequestManager implements RequestManager { private final Logger log; private final Optional autoCommitState; private final CoordinatorRequestManager coordinatorRequestManager; - private final GroupStateManager groupState; + private final GroupState groupState; public CommitRequestManager( final Time time, @@ -56,7 +56,7 @@ public CommitRequestManager( final SubscriptionState subscriptionState, final ConsumerConfig config, final CoordinatorRequestManager coordinatorRequestManager, - final GroupStateManager groupState) { + final GroupState groupState) { Objects.requireNonNull(coordinatorRequestManager, "Coordinator is needed upon committing offsets"); this.log = logContext.logger(getClass()); this.stagedCommits = new LinkedList<>(); @@ -88,6 +88,7 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { List unsentCommitRequests = stagedCommits.stream().map(StagedCommit::toUnsentRequest).collect(Collectors.toList()); + stagedCommits.clear(); return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.unmodifiableList(unsentCommitRequests)); } @@ -107,32 +108,35 @@ private void maybeAutoCommit(final long currentTimeMs) { } AutoCommitState autocommit = autoCommitState.get(); - if (!autocommit.canSendAutocommit(currentTimeMs)) { + if (!autocommit.canSendAutocommit()) { return; } Map allConsumedOffsets = subscriptionState.allConsumed(); log.debug("Auto-committing offsets {}", allConsumedOffsets); sendAutoCommit(allConsumedOffsets); - autocommit.reset(); + autocommit.resetTimer(); + } // Visible for testing CompletableFuture sendAutoCommit(final Map allConsumedOffsets) { - CompletableFuture future = this.add(allConsumedOffsets); - future.whenComplete((response, throwable) -> { - if (throwable == null) { - log.debug("Completed asynchronous auto-commit of offsets {}", allConsumedOffsets); - } - - if (throwable instanceof RetriableCommitFailedException) { - log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", allConsumedOffsets, - throwable); - } else { - log.warn("Asynchronous auto-commit of offsets {} failed: {}", allConsumedOffsets, - throwable.getMessage()); - } - }); + CompletableFuture future = this.add(allConsumedOffsets) + .whenComplete((response, throwable) -> { + if (throwable == null) { + log.debug("Completed asynchronous auto-commit of offsets {}", allConsumedOffsets); + } + // setting inflight commit to false upon completion + autoCommitState.get().setInflightCommitStatus(false); + }) + .exceptionally(t -> { + if (t instanceof RetriableCommitFailedException) { + log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", allConsumedOffsets, t); + } else { + log.warn("Asynchronous auto-commit of offsets {} failed: {}", allConsumedOffsets, t.getMessage()); + } + return null; + }); return future; } @@ -143,14 +147,14 @@ public void clientPoll(final long currentTimeMs) { private class StagedCommit { private final Map offsets; private final String groupId; - private final GroupStateManager.Generation generation; + private final GroupState.Generation generation; private final String groupInstanceId; private final NetworkClientDelegate.FutureCompletionHandler future; public StagedCommit(final Map offsets, final String groupId, final String groupInstanceId, - final GroupStateManager.Generation generation) { + final GroupState.Generation generation) { this.offsets = offsets; // if no callback is provided, DefaultOffsetCommitCallback will be used. this.future = new NetworkClientDelegate.FutureCompletionHandler(); @@ -201,6 +205,7 @@ public NetworkClientDelegate.UnsentRequest toUnsentRequest() { static class AutoCommitState { private final Timer timer; private final long autoCommitInterval; + private boolean hasInflightCommit; public AutoCommitState( final Time time, @@ -209,14 +214,18 @@ public AutoCommitState( this.timer = time.timer(autoCommitInterval); } - public boolean canSendAutocommit(final long currentTimeMs) { - return this.timer.isExpired(); + public boolean canSendAutocommit() { + return !hasInflightCommit && this.timer.isExpired(); } - public void reset() { + public void resetTimer() { this.timer.reset(autoCommitInterval); } + public void setInflightCommitStatus(final boolean hasInflightCommit) { + this.hasInflightCommit = hasInflightCommit; + } + public void ack(final long currentTimeMs) { this.timer.update(currentTimeMs); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java index b565b33e676b0..3820dfe639a54 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java @@ -61,7 +61,7 @@ public class DefaultBackgroundThread extends KafkaThread { private final ApplicationEventProcessor applicationEventProcessor; private final NetworkClientDelegate networkClientDelegate; private final ErrorEventHandler errorEventHandler; - private final GroupStateManager groupState; + private final GroupState groupState; private boolean running; private final Map> requestManagerRegistry; @@ -76,7 +76,7 @@ public class DefaultBackgroundThread extends KafkaThread { final ApplicationEventProcessor processor, final ConsumerMetadata metadata, final NetworkClientDelegate networkClient, - final GroupStateManager groupState, + final GroupState groupState, final CoordinatorRequestManager coordinatorManager, CommitRequestManager commitRequestManager) { super(BACKGROUND_THREAD_NAME, true); @@ -120,7 +120,7 @@ public DefaultBackgroundThread(final Time time, networkClient); this.running = true; this.errorEventHandler = new ErrorEventHandler(this.backgroundEventQueue); - this.groupState = new GroupStateManager(rebalanceConfig); + this.groupState = new GroupState(rebalanceConfig); this.requestManagerRegistry = Collections.unmodifiableMap(buildRequestManagerRegistry(logContext)); this.applicationEventProcessor = new ApplicationEventProcessor(backgroundEventQueue, requestManagerRegistry); } catch (final Exception e) { diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupStateManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupState.java similarity index 94% rename from clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupStateManager.java rename to clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupState.java index f7c0083c5137d..4e16055f6fdb9 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupStateManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupState.java @@ -23,17 +23,17 @@ import java.util.Objects; import java.util.Optional; -public class GroupStateManager { +public class GroupState { public final String groupId; public final Optional groupInstanceId; public Generation generation = Generation.NO_GENERATION; - public GroupStateManager(String groupId, Optional groupInstanceId) { + public GroupState(String groupId, Optional groupInstanceId) { this.groupId = groupId; this.groupInstanceId = groupInstanceId; } - public GroupStateManager(final GroupRebalanceConfig config) { + public GroupState(final GroupRebalanceConfig config) { this.groupId = config.groupId; this.groupInstanceId = config.groupInstanceId; } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java index 0d005170d5f70..62945bd46f10d 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java @@ -63,7 +63,7 @@ private boolean process(final NoopApplicationEvent event) { private boolean process(final PollApplicationEvent event) { Optional commitRequestManger = registry.get(RequestManager.Type.COMMIT); if (!commitRequestManger.isPresent()) { - return false; + return true; } CommitRequestManager manager = (CommitRequestManager) commitRequestManger.get(); diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java index 562126e972b0d..67c416a8a2fd8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java @@ -58,6 +58,7 @@ private Optional isValid(final Map @Override public String toString() { - return getClass() + "_" + this.offsets; + return "CommitApplicationEvent(" + + "offsets=" + offsets + ")"; } } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java index 6694919463abc..e7658ebe82db6 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java @@ -18,6 +18,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.utils.LogContext; @@ -38,7 +39,7 @@ public class CommitRequestManagerTest { private SubscriptionState subscriptionState; - private GroupStateManager groupStateManager; + private GroupState groupState; private LogContext logContext; private MockTime time; private CoordinatorRequestManager coordinatorRequestManager; @@ -50,7 +51,7 @@ public void setup() { this.time = new MockTime(0); this.subscriptionState = mock(SubscriptionState.class); this.coordinatorRequestManager = mock(CoordinatorRequestManager.class); - this.groupStateManager = new GroupStateManager("group-1", Optional.empty()); + this.groupState = new GroupState("group-1", Optional.empty()); this.props = new Properties(); this.props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100); @@ -89,6 +90,25 @@ public void testPollAndAutoCommit() { assertEquals(1, res.unsentRequests.size()); } + @Test + public void testAutocommitStateUponFailure() { + 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); + + // complete the autocommit request (exceptionally) + res.unsentRequests.get(0).future().completeExceptionally(new KafkaException("test exception")); + + // we can then autocommit again + commitRequestManger.clientPoll(time.milliseconds()); + res = commitRequestManger.poll(time.milliseconds()); + assertEquals(1, res.unsentRequests.size()); + } + @Test public void testAutoCommitFuture() { CommitRequestManager commitRequestManger = create(true, 100); @@ -104,6 +124,6 @@ private CommitRequestManager create(final boolean autoCommitEnabled, final long this.subscriptionState, new ConsumerConfig(props), this.coordinatorRequestManager, - this.groupStateManager); + this.groupState); } } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java index 73d29b364b3aa..dbc1753959083 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java @@ -62,7 +62,7 @@ public class DefaultBackgroundThreadTest { private CoordinatorRequestManager coordinatorManager; private ErrorEventHandler errorEventHandler; private int requestTimeoutMs = 500; - private GroupStateManager groupState; + private GroupState groupState; private CommitRequestManager commitManager; @BeforeEach @@ -84,7 +84,7 @@ public void setup() { Optional.empty(), 100, true); - this.groupState = new GroupStateManager(rebalanceConfig); + this.groupState = new GroupState(rebalanceConfig); this.commitManager = mock(CommitRequestManager.class); } From d98695f57bb921a47f344df764b35add67f7fd11 Mon Sep 17 00:00:00 2001 From: Philip Nee Date: Mon, 6 Feb 2023 13:16:24 -0800 Subject: [PATCH 5/6] ensure staged commit is purged (testing) --- .../consumer/internals/CommitRequestManager.java | 5 +++++ .../consumer/internals/CommitRequestManagerTest.java | 10 ++++++++++ 2 files changed, 15 insertions(+) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java index cb05959bdbc2c..00a722a9f9e7a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java @@ -144,6 +144,11 @@ public void clientPoll(final long currentTimeMs) { this.autoCommitState.ifPresent(t -> t.ack(currentTimeMs)); } + // Visible for testing + Queue stagedCommits() { + return this.stagedCommits; + } + private class StagedCommit { private final Map offsets; private final String groupId; diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java index e7658ebe82db6..95749bde194bd 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java @@ -34,6 +34,7 @@ import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -109,6 +110,15 @@ public void testAutocommitStateUponFailure() { assertEquals(1, res.unsentRequests.size()); } + @Test + public void testEnsureStagedCommitsPurgedAfterPoll() { + 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()); + } + @Test public void testAutoCommitFuture() { CommitRequestManager commitRequestManger = create(true, 100); From da218d05311241b8c0a66509805856375534876b Mon Sep 17 00:00:00 2001 From: Philip Nee Date: Mon, 6 Feb 2023 15:56:48 -0800 Subject: [PATCH 6/6] make commit manager final --- .../clients/consumer/internals/DefaultBackgroundThread.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java index 3820dfe639a54..62f555b356d37 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java @@ -78,7 +78,7 @@ public class DefaultBackgroundThread extends KafkaThread { final NetworkClientDelegate networkClient, final GroupState groupState, final CoordinatorRequestManager coordinatorManager, - CommitRequestManager commitRequestManager) { + final CommitRequestManager commitRequestManager) { super(BACKGROUND_THREAD_NAME, true); this.time = time; this.running = true;