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..00a722a9f9e7a --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java @@ -0,0 +1,238 @@ +/* + * 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.Objects; +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 GroupState groupState; + + public CommitRequestManager( + final Time time, + final LogContext logContext, + final SubscriptionState subscriptionState, + final ConsumerConfig config, + final CoordinatorRequestManager coordinatorRequestManager, + final GroupState 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)) { + 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; + } + + /** + * 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) { + maybeAutoCommit(currentTimeMs); + + if (stagedCommits.isEmpty()) { + return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, new ArrayList<>()); + } + + List unsentCommitRequests = + stagedCommits.stream().map(StagedCommit::toUnsentRequest).collect(Collectors.toList()); + stagedCommits.clear(); + 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()) { + return; + } + + Map allConsumedOffsets = subscriptionState.allConsumed(); + log.debug("Auto-committing offsets {}", allConsumedOffsets); + sendAutoCommit(allConsumedOffsets); + autocommit.resetTimer(); + + } + + // Visible for testing + CompletableFuture sendAutoCommit(final Map allConsumedOffsets) { + 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; + } + + 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; + 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 GroupState.Generation generation) { + this.offsets = offsets; + // if no callback is provided, DefaultOffsetCommitCallback will be used. + this.future = new NetworkClientDelegate.FutureCompletionHandler(); + this.groupId = groupId; + this.generation = generation; + this.groupInstanceId = groupInstanceId; + } + + public CompletableFuture future() { + return future.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(), + future); + } + } + + static class AutoCommitState { + private final Timer timer; + private final long autoCommitInterval; + private boolean hasInflightCommit; + + public AutoCommitState( + final Time time, + final long autoCommitInterval) { + this.autoCommitInterval = autoCommitInterval; + this.timer = time.timer(autoCommitInterval); + } + + public boolean canSendAutocommit() { + return !hasInflightCommit && this.timer.isExpired(); + } + + 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 481137e074786..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 @@ -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 GroupState 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 GroupState groupState, + final CoordinatorRequestManager coordinatorManager, + final 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 GroupState(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,31 +174,25 @@ 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(); - + drain(); 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 -> 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(); - Iterator iter = events.iterator(); - while (iter.hasNext()) { - ApplicationEvent event = iter.next(); - log.debug("processing application event: {}", event); + for (ApplicationEvent event : events) { + log.debug("Consuming application event: {}", event); consumeApplicationEvent(event); } } @@ -198,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/GroupState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupState.java new file mode 100644 index 0000000000000..4e16055f6fdb9 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupState.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 GroupState { + public final String groupId; + public final Optional groupInstanceId; + public Generation generation = Generation.NO_GENERATION; + + public GroupState(String groupId, Optional groupInstanceId) { + this.groupId = groupId; + this.groupInstanceId = groupInstanceId; + } + + public GroupState(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..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 @@ -16,22 +16,35 @@ */ package org.apache.kafka.clients.consumer.internals.events; +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; 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 +59,36 @@ 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 true; + } + + 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()) { + // 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()).whenComplete((r, e) -> { + if (e != null) { + event.future().completeExceptionally(e); + return; + } + event.future().complete(null); + }); + 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..67c416a8a2fd8 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitApplicationEvent.java @@ -0,0 +1,64 @@ +/* + * 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 "CommitApplicationEvent(" + + "offsets=" + 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..95749bde194bd --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java @@ -0,0 +1,139 @@ +/* + * 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.KafkaException; +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.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class CommitRequestManagerTest { + private SubscriptionState subscriptionState; + private GroupState groupState; + 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.groupState = new GroupState("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()); + } + + @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 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); + 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, + this.logContext, + this.subscriptionState, + new ConsumerConfig(props), + this.coordinatorRequestManager, + 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 7ee4ee6a1263d..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 @@ -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 GroupState 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 GroupState(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,7 +113,8 @@ public void testApplicationEvent() { @Test void testFindCoordinator() { DefaultBackgroundThread backgroundThread = mockBackgroundThread(); - when(this.coordinatorManager.poll(time.milliseconds())).thenReturn(mockPollResult()); + 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(networkClient, times(1)).poll(anyLong(), anyLong()); @@ -122,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( @@ -138,7 +153,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 +165,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))); } }