You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2023/12/21 14:53:11 UTC
(kafka) branch 3.7 updated: KAFKA-15456: Commit/Fetch error handling improvements and V9 support (#14557)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch 3.7
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.7 by this push:
new fe82ad813ad KAFKA-15456: Commit/Fetch error handling improvements and V9 support (#14557)
fe82ad813ad is described below
commit fe82ad813ad2e4ecec6590898b6ce90c51b75d04
Author: Lianet Magrans <98...@users.noreply.github.com>
AuthorDate: Thu Dec 21 09:48:34 2023 -0500
KAFKA-15456: Commit/Fetch error handling improvements and V9 support (#14557)
This patch includes multiple fixes for OffsetFetch and OffsetCommit related to error handling and retries, introducing support for new errors as defined in the requests version 9.
- Fixes for OffsetFetch and OffsetCommit error handling (retry when applicable, properly time-bound retries)
- Fixes for commitSync retried and commitAsync not retried
- Support for OffsetFetch and OffsetCommit v9, retried with backoff with the latest memberID/epoch if the member is still in the group, or failed if the member is not in the group anymore (left/failed/fenced).
- Minor refactoring for removing unneeded group state and generation info and other minor clean up.
Reviewers: Andrew Schofield <as...@confluent.io>, Kirk True <kt...@confluent.io>, Bruno Cadonna <ca...@apache.org>, Lucas Brutschy <lb...@confluent.io>, David Jacot <dj...@confluent.io>
---
.../clients/consumer/internals/AbstractFetch.java | 6 +-
.../consumer/internals/AsyncKafkaConsumer.java | 33 +-
.../consumer/internals/CommitRequestManager.java | 691 +++++++++++++++------
.../clients/consumer/internals/GroupState.java | 89 ---
.../consumer/internals/MemberStateListener.java | 37 ++
.../consumer/internals/MembershipManager.java | 6 +
.../consumer/internals/MembershipManagerImpl.java | 214 ++++---
.../consumer/internals/RequestManagers.java | 16 +-
.../events/ApplicationEventProcessor.java | 30 +-
.../internals/events/CommitApplicationEvent.java | 25 +-
.../FetchCommittedOffsetsApplicationEvent.java | 17 +-
.../consumer/internals/AsyncKafkaConsumerTest.java | 18 +-
.../internals/CommitRequestManagerTest.java | 539 +++++++++++++---
.../internals/ConsumerNetworkThreadTest.java | 11 +-
.../consumer/internals/ConsumerTestBuilder.java | 48 +-
.../internals/HeartbeatRequestManagerTest.java | 3 +-
.../internals/MembershipManagerImplTest.java | 202 ++++--
.../ApplicationEventProcessorTest.java | 25 +-
18 files changed, 1441 insertions(+), 569 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
index 07c82907fef..1d0c9b09f61 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
@@ -16,14 +16,12 @@
*/
package org.apache.kafka.clients.consumer.internals;
-import java.util.List;
-import java.util.stream.Collectors;
import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.ClientResponse;
import org.apache.kafka.clients.FetchSessionHandler;
import org.apache.kafka.clients.KafkaClient;
-import org.apache.kafka.clients.NetworkClientUtils;
import org.apache.kafka.clients.Metadata;
+import org.apache.kafka.clients.NetworkClientUtils;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
@@ -47,10 +45,12 @@ import java.io.Closeable;
import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.function.Predicate;
+import java.util.stream.Collectors;
import static org.apache.kafka.clients.consumer.internals.FetchUtils.requestMetadataUpdate;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index 6f0eaef4eb5..91f9718b285 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -36,7 +36,6 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata;
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.RetriableCommitFailedException;
import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent;
import org.apache.kafka.clients.consumer.internals.events.ApplicationEventHandler;
import org.apache.kafka.clients.consumer.internals.events.ApplicationEventProcessor;
@@ -71,7 +70,6 @@ import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.FencedInstanceIdException;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.InvalidGroupIdException;
-import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.metrics.Metrics;
@@ -755,7 +753,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
public void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback) {
acquireAndEnsureOpen();
try {
- CompletableFuture<Void> future = commit(offsets, false);
+ // Commit without timer to indicate that the commit should be triggered without
+ // waiting for a response.
+ CompletableFuture<Void> future = commit(offsets, false, Optional.empty());
future.whenComplete((r, t) -> {
if (callback == null) {
if (t != null) {
@@ -772,7 +772,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
}
// Visible for testing
- CompletableFuture<Void> commit(final Map<TopicPartition, OffsetAndMetadata> offsets, final boolean isWakeupable) {
+ CompletableFuture<Void> commit(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final boolean isWakeupable,
+ final Optional<Long> retryTimeoutMs) {
maybeInvokeCommitCallbacks();
maybeThrowFencedInstanceException();
maybeThrowInvalidGroupIdException();
@@ -784,7 +786,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
return CompletableFuture.completedFuture(null);
}
- final CommitApplicationEvent commitEvent = new CommitApplicationEvent(offsets);
+ final CommitApplicationEvent commitEvent = new CommitApplicationEvent(offsets, retryTimeoutMs);
if (isWakeupable) {
// the task can only be woken up if the top level API call is commitSync
wakeupTrigger.setActiveTask(commitEvent.future());
@@ -921,7 +923,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
return Collections.emptyMap();
}
- final FetchCommittedOffsetsApplicationEvent event = new FetchCommittedOffsetsApplicationEvent(partitions);
+ final FetchCommittedOffsetsApplicationEvent event = new FetchCommittedOffsetsApplicationEvent(
+ partitions,
+ timeout.toMillis());
wakeupTrigger.setActiveTask(event.future());
try {
final Map<TopicPartition, OffsetAndMetadata> committedOffsets = applicationEventHandler.addAndGet(event,
@@ -1331,8 +1335,11 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
acquireAndEnsureOpen();
long commitStart = time.nanoseconds();
try {
- CompletableFuture<Void> commitFuture = commit(offsets, true);
- ConsumerUtils.getResult(commitFuture, time.timer(timeout));
+ Timer requestTimer = time.timer(timeout.toMillis());
+ // Commit with a timer to control how long the request should be retried until it
+ // gets a successful response or non-retriable error.
+ CompletableFuture<Void> commitFuture = commit(offsets, true, Optional.of(timeout.toMillis()));
+ ConsumerUtils.getResult(commitFuture, requestTimer);
} finally {
wakeupTrigger.clearTask();
kafkaConsumerMetrics.recordCommitSync(time.nanoseconds() - commitStart);
@@ -1596,7 +1603,10 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
log.debug("Refreshing committed offsets for partitions {}", initializingPartitions);
try {
- final FetchCommittedOffsetsApplicationEvent event = new FetchCommittedOffsetsApplicationEvent(initializingPartitions);
+ final FetchCommittedOffsetsApplicationEvent event =
+ new FetchCommittedOffsetsApplicationEvent(
+ initializingPartitions,
+ timer.remainingMs());
final Map<TopicPartition, OffsetAndMetadata> offsets = applicationEventHandler.addAndGet(event, timer);
refreshCommittedOffsets(offsets, metadata, subscriptions);
return true;
@@ -1945,11 +1955,6 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
}
public void invoke() {
- if (exception instanceof RetriableException) {
- callback.onComplete(offsets, new RetriableCommitFailedException(exception));
- return;
- }
-
if (exception instanceof FencedInstanceIdException)
isFenced = true;
callback.onComplete(offsets, exception);
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 6316ead7934..142c81d7b6b 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
@@ -21,12 +21,12 @@ import org.apache.kafka.clients.consumer.CommitFailedException;
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.clients.consumer.internals.events.BackgroundEventHandler;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.message.OffsetCommitRequestData;
import org.apache.kafka.common.message.OffsetCommitResponseData;
@@ -61,20 +61,17 @@ import java.util.stream.Collectors;
import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED;
import static org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult.EMPTY;
import static org.apache.kafka.common.protocol.Errors.COORDINATOR_LOAD_IN_PROGRESS;
-import static org.apache.kafka.common.protocol.Errors.COORDINATOR_NOT_AVAILABLE;
-import static org.apache.kafka.common.protocol.Errors.NOT_COORDINATOR;
-import static org.apache.kafka.common.protocol.Errors.REQUEST_TIMED_OUT;
-public class CommitRequestManager implements RequestManager {
+public class CommitRequestManager implements RequestManager, MemberStateListener {
private final SubscriptionState subscriptions;
private final LogContext logContext;
private final Logger log;
private final Optional<AutoCommitState> autoCommitState;
- private final BackgroundEventHandler backgroundEventHandler;
private final CoordinatorRequestManager coordinatorRequestManager;
- private final GroupState groupState;
private final long retryBackoffMs;
+ private final String groupId;
+ private final Optional<String> groupInstanceId;
private final long retryBackoffMaxMs;
// For testing only
private final OptionalDouble jitter;
@@ -82,33 +79,25 @@ public class CommitRequestManager implements RequestManager {
final PendingRequests pendingRequests;
private boolean closing = false;
+ /**
+ * Latest member ID and epoch received via the {@link #onMemberEpochUpdated(Optional, Optional)},
+ * to be included in the OffsetFetch and OffsetCommit requests if present. This will have
+ * the latest values received from the broker, or empty of the member is not part of the
+ * group anymore.
+ */
+ private final MemberInfo memberInfo;
+
public CommitRequestManager(
final Time time,
final LogContext logContext,
final SubscriptionState subscriptions,
final ConsumerConfig config,
final CoordinatorRequestManager coordinatorRequestManager,
- final BackgroundEventHandler backgroundEventHandler,
- final GroupState groupState) {
- Objects.requireNonNull(coordinatorRequestManager, "Coordinator is needed upon committing offsets");
- this.logContext = logContext;
- this.log = logContext.logger(getClass());
- this.pendingRequests = new PendingRequests();
- 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.backgroundEventHandler = backgroundEventHandler;
- this.coordinatorRequestManager = coordinatorRequestManager;
- this.groupState = groupState;
- this.subscriptions = subscriptions;
- this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
- this.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
- this.jitter = OptionalDouble.empty();
- this.throwOnFetchStableOffsetUnsupported = config.getBoolean(THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED);
+ final String groupId,
+ final Optional<String> groupInstanceId) {
+ this(time, logContext, subscriptions, config, coordinatorRequestManager, groupId,
+ groupInstanceId, config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG),
+ config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG), OptionalDouble.empty());
}
// Visible for testing
@@ -118,11 +107,11 @@ public class CommitRequestManager implements RequestManager {
final SubscriptionState subscriptions,
final ConsumerConfig config,
final CoordinatorRequestManager coordinatorRequestManager,
- final BackgroundEventHandler backgroundEventHandler,
- final GroupState groupState,
+ final String groupId,
+ final Optional<String> groupInstanceId,
final long retryBackoffMs,
final long retryBackoffMaxMs,
- final double jitter) {
+ final OptionalDouble jitter) {
Objects.requireNonNull(coordinatorRequestManager, "Coordinator is needed upon committing offsets");
this.logContext = logContext;
this.log = logContext.logger(getClass());
@@ -134,14 +123,15 @@ public class CommitRequestManager implements RequestManager {
} else {
this.autoCommitState = Optional.empty();
}
- this.backgroundEventHandler = backgroundEventHandler;
this.coordinatorRequestManager = coordinatorRequestManager;
- this.groupState = groupState;
+ this.groupId = groupId;
+ this.groupInstanceId = groupInstanceId;
this.subscriptions = subscriptions;
this.retryBackoffMs = retryBackoffMs;
this.retryBackoffMaxMs = retryBackoffMaxMs;
- this.jitter = OptionalDouble.of(jitter);
+ this.jitter = jitter;
this.throwOnFetchStableOffsetUnsupported = config.getBoolean(THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED);
+ this.memberInfo = new MemberInfo();
}
/**
@@ -158,7 +148,7 @@ public class CommitRequestManager implements RequestManager {
return drainPendingOffsetCommitRequests();
}
- maybeAutoCommitAllConsumed();
+ maybeAutoCommitAllConsumedAsync();
if (!pendingRequests.hasUnsentRequests())
return EMPTY;
@@ -199,21 +189,36 @@ public class CommitRequestManager implements RequestManager {
* request if there is no other commit request already in-flight, and if the commit interval
* has elapsed.
*
- * @param offsets Offsets to commit
+ * @param offsets Offsets to commit
+ * @param expirationTimeMs Time until which the request will continue to be retried if it
+ * fails with a retriable error. If not present, the request will be
+ * sent but not retried.
+ * @param checkInterval True if the auto-commit interval expiration should be checked for
+ * sending a request. If true, the request will be sent only if the
+ * auto-commit interval has expired. Pass false to
+ * send the auto-commit request regardless of the interval (ex.
+ * auto-commit before rebalance).
+ * @param retryOnStaleEpoch True if the request should be retried in case it fails with
+ * {@link Errors#STALE_MEMBER_EPOCH}.
* @return Future that will complete when a response is received for the request, or a
* completed future if no request is generated.
*/
- public CompletableFuture<Void> maybeAutoCommit(final Map<TopicPartition, OffsetAndMetadata> offsets) {
- if (!canAutoCommit()) {
+ private CompletableFuture<Void> maybeAutoCommit(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final Optional<Long> expirationTimeMs,
+ boolean checkInterval,
+ boolean retryOnStaleEpoch) {
+ if (!autoCommitEnabled()) {
+ log.debug("Skipping auto-commit because auto-commit config is not enabled.");
return CompletableFuture.completedFuture(null);
}
AutoCommitState autocommit = autoCommitState.get();
- if (!autocommit.shouldAutoCommit()) {
+ if (checkInterval && !autocommit.shouldAutoCommit()) {
return CompletableFuture.completedFuture(null);
}
- CompletableFuture<Void> result = sendAutoCommit(offsets);
+ CompletableFuture<Void> result = addOffsetCommitRequest(offsets, expirationTimeMs, retryOnStaleEpoch)
+ .whenComplete(autoCommitCallback(offsets));
autocommit.resetTimer();
autocommit.setInflightCommitStatus(true);
return result;
@@ -221,40 +226,45 @@ public class CommitRequestManager implements RequestManager {
/**
* If auto-commit is enabled, this will generate a commit offsets request for all assigned
- * partitions and their current positions.
+ * partitions and their current positions. Note on auto-commit timers: this will reset the
+ * auto-commit timer to the interval before issuing the async commit, and when the async commit
+ * completes, it will reset the auto-commit timer with the exponential backoff if the request
+ * failed with a retriable error.
*
* @return Future that will complete when a response is received for the request, or a
* completed future if no request is generated.
*/
- public CompletableFuture<Void> maybeAutoCommitAllConsumed() {
- return maybeAutoCommit(subscriptions.allConsumed());
- }
-
- boolean canAutoCommit() {
- return autoCommitState.isPresent() && !subscriptions.allConsumed().isEmpty();
- }
+ public CompletableFuture<Void> maybeAutoCommitAllConsumedAsync() {
+ if (!autoCommitEnabled()) {
+ // Early return to ensure that no action/logging is performed.
+ return CompletableFuture.completedFuture(null);
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsets = subscriptions.allConsumed();
+ CompletableFuture<Void> result = maybeAutoCommit(offsets, Optional.empty(), true, true);
+ result.whenComplete((__, error) -> {
+ if (error != null) {
+ if (error instanceof RetriableCommitFailedException) {
+ log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error.", offsets, error);
+ resetAutoCommitTimer(retryBackoffMs);
+ } else {
+ log.warn("Asynchronous auto-commit of offsets {} failed: {}", offsets, error.getMessage());
+ }
+ } else {
+ log.debug("Completed asynchronous auto-commit of offsets {}", offsets);
+ }
+ });
- /**
- * Updates the member ID and epoch upon receiving ConsumerGroupHeartbeatResponse.
- */
- void updateMemberInformation(String memberId, int memberEpoch) {
- groupState.generation = new GroupState.Generation(memberEpoch, memberId, null);
+ return result;
}
/**
- * Returns an OffsetCommitRequest of all assigned topicPartitions and their current positions.
+ * Commit consumed offsets if auto-commit is enabled. Retry while the timer is not expired,
+ * until the request succeeds or fails with a fatal error.
*/
- NetworkClientDelegate.UnsentRequest createCommitAllConsumedRequest() {
- Map<TopicPartition, OffsetAndMetadata> offsets = subscriptions.allConsumed();
- OffsetCommitRequestState request = pendingRequests.createOffsetCommitRequest(offsets, jitter);
- log.debug("Sending synchronous auto-commit of offsets {}", offsets);
- request.future.whenComplete(autoCommitCallback(subscriptions.allConsumed()));
- return request.toUnsentRequest();
- }
-
- private CompletableFuture<Void> sendAutoCommit(final Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets) {
- log.debug("Enqueuing autocommit offsets: {}", allConsumedOffsets);
- return addOffsetCommitRequest(allConsumedOffsets).whenComplete(autoCommitCallback(allConsumedOffsets));
+ public CompletableFuture<Void> maybeAutoCommitAllConsumedNow(
+ final Optional<Long> expirationTimeMs,
+ final boolean retryOnStaleEpoch) {
+ return maybeAutoCommit(subscriptions.allConsumed(), expirationTimeMs, false, retryOnStaleEpoch);
}
private BiConsumer<? super Void, ? super Throwable> autoCommitCallback(final Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets) {
@@ -275,16 +285,24 @@ public class CommitRequestManager implements RequestManager {
* Handles {@link org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent}. It creates an
* {@link OffsetCommitRequestState} and enqueue it to send later.
*/
- public CompletableFuture<Void> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
- return pendingRequests.addOffsetCommitRequest(offsets).future();
+ public CompletableFuture<Void> addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final Optional<Long> expirationTimeMs,
+ final boolean retryOnStaleEpoch) {
+ if (offsets.isEmpty()) {
+ log.debug("Skipping commit of empty offsets");
+ return CompletableFuture.completedFuture(null);
+ }
+ return pendingRequests.addOffsetCommitRequest(offsets, expirationTimeMs, retryOnStaleEpoch).future;
}
/**
* Handles {@link org.apache.kafka.clients.consumer.internals.events.FetchCommittedOffsetsApplicationEvent}. It creates an
* {@link OffsetFetchRequestState} and enqueue it to send later.
*/
- public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
- return pendingRequests.addOffsetFetchRequest(partitions);
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(
+ final Set<TopicPartition> partitions,
+ final long expirationTimeMs) {
+ return pendingRequests.addOffsetFetchRequest(partitions, expirationTimeMs);
}
public void updateAutoCommitTimer(final long currentTimeMs) {
@@ -306,6 +324,18 @@ public class CommitRequestManager implements RequestManager {
}
}
+ /**
+ * Update latest member ID and epoch used by the member.
+ *
+ * @param memberEpoch New member epoch received. To be included in the new request.
+ * @param memberId Current member ID. To be included in the new request.
+ */
+ @Override
+ public void onMemberEpochUpdated(Optional<Integer> memberEpoch, Optional<String> memberId) {
+ memberInfo.memberId = memberId;
+ memberInfo.memberEpoch = memberEpoch;
+ }
+
/**
* @return True if auto-commit is enabled as defined in the config {@link ConsumerConfig#ENABLE_AUTO_COMMIT_CONFIG}
*/
@@ -314,13 +344,22 @@ public class CommitRequestManager implements RequestManager {
}
/**
- * Reset the auto-commit timer so that the next auto-commit is sent out on the interval
- * starting from now. If auto-commit is not enabled this will perform no action.
+ * Reset the auto-commit timer to the auto-commit interval, so that the next auto-commit is
+ * sent out on the interval starting from now. If auto-commit is not enabled this will
+ * perform no action.
*/
public void resetAutoCommitTimer() {
autoCommitState.ifPresent(AutoCommitState::resetTimer);
}
+ /**
+ * Reset the auto-commit timer to the provided time (backoff), so that the next auto-commit is
+ * sent out then. If auto-commit is not enabled this will perform no action.
+ */
+ public void resetAutoCommitTimer(long retryBackoffMs) {
+ autoCommitState.ifPresent(s -> s.resetTimer(retryBackoffMs));
+ }
+
/**
* Drains the inflight offsetCommits during shutdown because we want to make sure all pending commits are sent
* before closing.
@@ -332,40 +371,54 @@ public class CommitRequestManager implements RequestManager {
return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, requests);
}
- private class OffsetCommitRequestState extends RequestState {
+ private class OffsetCommitRequestState extends RetriableRequestState {
private final Map<TopicPartition, OffsetAndMetadata> offsets;
private final String groupId;
- private final GroupState.Generation generation;
- private final String groupInstanceId;
+ private final Optional<String> groupInstanceId;
+
private final CompletableFuture<Void> future;
+ /**
+ * Time until which the request should be retried if it fails with retriable
+ * errors. If not present, the request is triggered without waiting for a response or
+ * retrying.
+ */
+ private final Optional<Long> expirationTimeMs;
+
OffsetCommitRequestState(final Map<TopicPartition, OffsetAndMetadata> offsets,
final String groupId,
- final String groupInstanceId,
- final GroupState.Generation generation,
+ final Optional<String> groupInstanceId,
+ final Optional<Long> expirationTimeMs,
final long retryBackoffMs,
- final long retryBackoffMaxMs) {
- super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, retryBackoffMaxMs);
+ final long retryBackoffMaxMs,
+ final MemberInfo memberInfo,
+ final boolean retryOnStaleEpoch) {
+ super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs, memberInfo, retryOnStaleEpoch);
this.offsets = offsets;
- this.future = new CompletableFuture<>();
this.groupId = groupId;
- this.generation = generation;
this.groupInstanceId = groupInstanceId;
+ this.future = new CompletableFuture<>();
+ this.expirationTimeMs = expirationTimeMs;
}
+ // Visible for testing
OffsetCommitRequestState(final Map<TopicPartition, OffsetAndMetadata> offsets,
final String groupId,
- final String groupInstanceId,
- final GroupState.Generation generation,
+ final Optional<String> groupInstanceId,
+ final Optional<Long> expirationTimeMs,
final long retryBackoffMs,
final long retryBackoffMaxMs,
- final double jitter) {
- super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, 2, retryBackoffMaxMs, jitter);
+ final double jitter,
+ final MemberInfo memberInfo,
+ final boolean retryOnStaleEpoch) {
+ super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, 2,
+ retryBackoffMaxMs, jitter, memberInfo, retryOnStaleEpoch);
this.offsets = offsets;
- this.future = new CompletableFuture<>();
this.groupId = groupId;
- this.generation = generation;
this.groupInstanceId = groupInstanceId;
+ this.future = new CompletableFuture<>();
+ this.expirationTimeMs = expirationTimeMs;
}
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
@@ -389,13 +442,19 @@ public class CommitRequestManager implements RequestManager {
requestTopicDataMap.put(topicPartition.topic(), topic);
}
- OffsetCommitRequest.Builder builder = new OffsetCommitRequest.Builder(
- new OffsetCommitRequestData()
+ OffsetCommitRequestData data = new OffsetCommitRequestData()
.setGroupId(this.groupId)
- .setGenerationIdOrMemberEpoch(generation.generationId)
- .setMemberId(generation.memberId)
- .setGroupInstanceId(groupInstanceId)
- .setTopics(new ArrayList<>(requestTopicDataMap.values())));
+ .setGroupInstanceId(groupInstanceId.orElse(null))
+ .setTopics(new ArrayList<>(requestTopicDataMap.values()));
+ if (memberInfo.memberId.isPresent()) {
+ data = data.setMemberId(memberInfo.memberId.get());
+ }
+ if (memberInfo.memberEpoch.isPresent()) {
+ data = data.setGenerationIdOrMemberEpoch(memberInfo.memberEpoch.get());
+ }
+
+ OffsetCommitRequest.Builder builder = new OffsetCommitRequest.Builder(data);
+
NetworkClientDelegate.UnsentRequest resp = new NetworkClientDelegate.UnsentRequest(
builder,
coordinatorRequestManager.coordinator());
@@ -403,9 +462,17 @@ public class CommitRequestManager implements RequestManager {
(response, throwable) -> {
try {
if (throwable == null) {
+ log.debug("OffsetCommit response received for offsets {} ", offsets);
onResponse(response);
} else {
- onError(throwable, resp.handler().completionTimeMs());
+ log.debug("OffsetCommit completed with error for offsets {}", offsets, throwable);
+ long currentTimeMs = resp.handler().completionTimeMs();
+ handleCoordinatorDisconnect(throwable, currentTimeMs);
+ if (throwable instanceof RetriableException) {
+ maybeRetry(currentTimeMs, throwable);
+ } else {
+ future.completeExceptionally(throwable);
+ }
}
} catch (Throwable t) {
log.error("Unexpected error when completing offset commit: {}", this, t);
@@ -415,19 +482,8 @@ public class CommitRequestManager implements RequestManager {
return resp;
}
- public CompletableFuture<Void> future() {
- return future;
- }
-
- public void onError(final Throwable exception, final long currentTimeMs) {
- if (exception instanceof RetriableException) {
- handleCoordinatorDisconnect(exception, currentTimeMs);
- retry(currentTimeMs);
- }
- }
-
public void onResponse(final ClientResponse response) {
- long responseTime = response.receivedTimeMs();
+ long currentTimeMs = response.receivedTimeMs();
OffsetCommitResponse commitResponse = (OffsetCommitResponse) response.responseBody();
Set<String> unauthorizedTopics = new HashSet<>();
for (OffsetCommitResponseData.OffsetCommitResponseTopic topic : commitResponse.data().topics()) {
@@ -437,21 +493,45 @@ public class CommitRequestManager implements RequestManager {
long offset = offsetAndMetadata.offset();
Errors error = Errors.forCode(partition.errorCode());
if (error == Errors.NONE) {
- log.debug("OffsetCommit {} for partition {}", offset, tp);
+ log.debug("OffsetCommit completed successfully for offset {} partition {}", offset, tp);
continue;
}
- if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
+ if (error == Errors.COORDINATOR_NOT_AVAILABLE ||
+ error == Errors.NOT_COORDINATOR ||
+ error == Errors.REQUEST_TIMED_OUT) {
+ coordinatorRequestManager.markCoordinatorUnknown(error.message(), currentTimeMs);
+ maybeRetry(currentTimeMs, error.exception());
+ return;
+ } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS ||
+ error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+ // just retry
+ maybeRetry(currentTimeMs, error.exception());
+ return;
+ } else if (error == Errors.UNKNOWN_MEMBER_ID) {
+ log.error("OffsetCommit failed with {} on partition {} for offset {}",
+ error, tp, offset);
+ future.completeExceptionally(new CommitFailedException("OffsetCommit " +
+ "failed with unknown member ID. " + error.message()));
+ return;
+ } else if (error == Errors.STALE_MEMBER_EPOCH) {
+ if (maybeRetryWithNewMemberEpoch(currentTimeMs, error)) {
+ log.debug("OffsetCommit failed with {} and will be retried with the " +
+ "latest member ID and epoch.", error);
+ return;
+ }
+ future.completeExceptionally(commitExceptionForStaleMemberEpoch());
+ } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
// Collect all unauthorized topics before failing
unauthorizedTopics.add(tp.topic());
- } else if (error.exception() instanceof RetriableException) {
- log.warn("OffsetCommit failed on partition {} at offset {}: {}", tp, offset, error.message());
- handleRetriableError(error, response);
- retry(responseTime);
- return;
} else {
- log.error("OffsetCommit failed on partition {} at offset {}: {}", tp, offset, error.message());
- handleFatalError(error);
+ log.error("OffsetCommit failed on partition {} for offset {}: {}",
+ tp, offset, error.message());
+ if (error.exception() instanceof RetriableException) {
+ maybeRetry(currentTimeMs, error.exception());
+ return;
+ }
+ future.completeExceptionally(error.exception());
return;
}
}
@@ -465,86 +545,203 @@ public class CommitRequestManager implements RequestManager {
}
}
- private void handleRetriableError(Errors error, ClientResponse response) {
- if (error == COORDINATOR_NOT_AVAILABLE ||
- error == NOT_COORDINATOR ||
- error == REQUEST_TIMED_OUT) {
- coordinatorRequestManager.markCoordinatorUnknown(error.message(), response.receivedTimeMs());
+ /**
+ * Enqueue the request to be retried with exponential backoff, if the request allows
+ * retries and the timer has not expired. Complete the request future exceptionally if
+ * the request won't be retried.
+ */
+ @Override
+ void maybeRetry(long currentTimeMs, Throwable throwable) {
+ if (!allowsRetries()) {
+ // Fail requests that do not allow retries (async requests), making sure to
+ // propagate a RetriableCommitException if the failure is retriable.
+ future.completeExceptionally(commitExceptionForRetriableError(throwable));
+ return;
+ }
+ if (isExpired(currentTimeMs)) {
+ // Fail requests that allowed retries (sync requests), but expired.
+ future.completeExceptionally(throwable);
+ return;
}
- }
- private void retry(final long currentTimeMs) {
+ // Enqueue request to be retried with backoff. Note that this maintains the same
+ // timer of the initial request, so all the retries are time-bounded.
onFailedAttempt(currentTimeMs);
pendingRequests.addOffsetCommitRequest(this);
}
- private void handleFatalError(final Errors error) {
- switch (error) {
- case GROUP_AUTHORIZATION_FAILED:
- future.completeExceptionally(GroupAuthorizationException.forGroupId(groupId));
- break;
- case OFFSET_METADATA_TOO_LARGE:
- case INVALID_COMMIT_OFFSET_SIZE:
- future.completeExceptionally(error.exception());
- break;
- case FENCED_INSTANCE_ID:
- log.info("OffsetCommit failed due to group instance id {} fenced: {}", groupInstanceId, error.message());
- future.completeExceptionally(new CommitFailedException());
- break;
- case UNKNOWN_MEMBER_ID:
- log.info("OffsetCommit failed due to unknown member id memberId {}: {}", null, error.message());
- future.completeExceptionally(error.exception());
- break;
- default:
- future.completeExceptionally(new KafkaException("Unexpected error in commit: " + error.message()));
- break;
+ private boolean isExpired(final long currentTimeMs) {
+ return expirationTimeMs.isPresent() && expirationTimeMs.get() <= currentTimeMs;
+ }
+
+ /**
+ * @return True if the requests allows to be retried (sync requests that provide an
+ * expiration time to bound the retries). False if the request does not allow to be
+ * retried on RetriableErrors (async requests that does not provide an expiration time
+ * for retries)
+ */
+ private boolean allowsRetries() {
+ return expirationTimeMs.isPresent();
+ }
+
+ /**
+ * Complete the request future with a TimeoutException if the request expired. No action
+ * taken if the request is still active.
+ *
+ * @return True if the request expired.
+ */
+ private boolean maybeExpire(final long currentTimeMs) {
+ if (isExpired(currentTimeMs)) {
+ future.completeExceptionally(new TimeoutException("OffsetCommit could not complete " +
+ "before timeout expired."));
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * @return A RetriableCommitFailedException for async commit requests if the original
+ * Exception was a RetriableException. Return the original one in any other case.
+ */
+ private Throwable commitExceptionForRetriableError(Throwable throwable) {
+ if (!allowsRetries() && throwable instanceof RetriableException)
+ return new RetriableCommitFailedException(throwable);
+ return throwable;
+ }
+
+ private Throwable commitExceptionForStaleMemberEpoch() {
+ if (retryOnStaleEpoch)
+ return new RetriableCommitFailedException(Errors.STALE_MEMBER_EPOCH.exception());
+ return new CommitFailedException("OffsetCommit failed with stale member epoch." + Errors.STALE_MEMBER_EPOCH.message());
+ }
+ }
+
+ /**
+ * Represents a request that can be retried or aborted, based on member ID and epoch
+ * information.
+ */
+ abstract static class RetriableRequestState extends RequestState {
+
+ /**
+ * Member info (ID and epoch) to be included in the request if present.
+ */
+ final MemberInfo memberInfo;
+
+ /**
+ * True if the request should be retried if it fails with {@link Errors#STALE_MEMBER_EPOCH}.
+ */
+ boolean retryOnStaleEpoch;
+
+ RetriableRequestState(LogContext logContext, String owner, long retryBackoffMs,
+ long retryBackoffMaxMs, MemberInfo memberInfo, boolean retryOnStaleEpoch) {
+ super(logContext, owner, retryBackoffMs, retryBackoffMaxMs);
+ this.memberInfo = memberInfo;
+ this.retryOnStaleEpoch = retryOnStaleEpoch;
+ }
+
+ // Visible for testing
+ RetriableRequestState(LogContext logContext, String owner, long retryBackoffMs, int retryBackoffExpBase,
+ long retryBackoffMaxMs, double jitter, MemberInfo memberInfo,
+ boolean retryOnStaleEpoch) {
+ super(logContext, owner, retryBackoffMs, retryBackoffExpBase, retryBackoffMaxMs, jitter);
+ this.memberInfo = memberInfo;
+ this.retryOnStaleEpoch = retryOnStaleEpoch;
+ }
+
+ /**
+ * Retry with backoff if the request failed with {@link Errors#STALE_MEMBER_EPOCH} and
+ * the member has valid epoch.
+ *
+ * @return True if the request has been enqueued to be retried with the latest member ID
+ * and epoch.
+ */
+ boolean maybeRetryWithNewMemberEpoch(long currentTimeMs, Errors responseError) {
+ if (retryOnStaleEpoch && memberInfo.memberEpoch.isPresent()) {
+ // Request failed with invalid epoch, but the member has a valid one, so
+ // retry the request with the latest ID/epoch.
+ maybeRetry(currentTimeMs, responseError.exception());
+ return true;
}
+ return false;
}
+
+ abstract void maybeRetry(long currentTimeMs, Throwable throwable);
}
- class OffsetFetchRequestState extends RequestState {
+ class OffsetFetchRequestState extends RetriableRequestState {
+
+ /**
+ * Partitions to get committed offsets for.
+ */
public final Set<TopicPartition> requestedPartitions;
- public final GroupState.Generation requestedGeneration;
+
private final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+
+ /**
+ * Time until which the request should be retried if it fails with retriable errors.
+ */
+ private final long expirationTimeMs;
+
public OffsetFetchRequestState(final Set<TopicPartition> partitions,
- final GroupState.Generation generation,
final long retryBackoffMs,
- final long retryBackoffMaxMs) {
- super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, retryBackoffMaxMs);
+ final long retryBackoffMaxMs,
+ final long expirationTimeMs,
+ final MemberInfo memberInfo) {
+ super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs,
+ retryBackoffMaxMs, memberInfo, true);
this.requestedPartitions = partitions;
- this.requestedGeneration = generation;
this.future = new CompletableFuture<>();
+ this.expirationTimeMs = expirationTimeMs;
}
public OffsetFetchRequestState(final Set<TopicPartition> partitions,
- final GroupState.Generation generation,
final long retryBackoffMs,
final long retryBackoffMaxMs,
- final double jitter) {
- super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, 2, retryBackoffMaxMs, jitter);
+ final long expirationTimeMs,
+ final double jitter,
+ final MemberInfo memberInfo) {
+ super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, 2,
+ retryBackoffMaxMs, jitter, memberInfo, true);
this.requestedPartitions = partitions;
- this.requestedGeneration = generation;
this.future = new CompletableFuture<>();
+ this.expirationTimeMs = expirationTimeMs;
}
public boolean sameRequest(final OffsetFetchRequestState request) {
- return Objects.equals(requestedGeneration, request.requestedGeneration) && requestedPartitions.equals(request.requestedPartitions);
+ return requestedPartitions.equals(request.requestedPartitions);
}
public NetworkClientDelegate.UnsentRequest toUnsentRequest() {
- OffsetFetchRequest.Builder builder = new OffsetFetchRequest.Builder(
- groupState.groupId,
- true,
- new ArrayList<>(this.requestedPartitions),
- throwOnFetchStableOffsetUnsupported);
+
+ OffsetFetchRequest.Builder builder;
+ if (memberInfo.memberId.isPresent() && memberInfo.memberEpoch.isPresent()) {
+ builder = new OffsetFetchRequest.Builder(
+ groupId,
+ memberInfo.memberId.get(),
+ memberInfo.memberEpoch.get(),
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ } else {
+ // Building request without passing member ID/epoch to leave the logic to choose
+ // default values when not present on the request builder.
+ builder = new OffsetFetchRequest.Builder(
+ groupId,
+ true,
+ new ArrayList<>(this.requestedPartitions),
+ throwOnFetchStableOffsetUnsupported);
+ }
return new NetworkClientDelegate.UnsentRequest(builder, coordinatorRequestManager.coordinator())
- .whenComplete((r, t) -> onResponse(r.receivedTimeMs(), (OffsetFetchResponse) r.responseBody()));
+ .whenComplete((r, t) -> onResponse(r.receivedTimeMs(), (OffsetFetchResponse) r.responseBody()));
}
+ /**
+ * Handle request responses, including successful and failed.
+ */
public void onResponse(
- final long currentTimeMs,
- final OffsetFetchResponse response) {
- Errors responseError = response.groupLevelError(groupState.groupId);
+ final long currentTimeMs,
+ final OffsetFetchResponse response) {
+ Errors responseError = response.groupLevelError(groupId);
if (responseError != Errors.NONE) {
onFailure(currentTimeMs, responseError);
return;
@@ -552,33 +749,81 @@ public class CommitRequestManager implements RequestManager {
onSuccess(currentTimeMs, response);
}
+ /**
+ * Handle failed responses. This will retry if the error is retriable, or complete the
+ * result future exceptionally in the case of non-recoverable or unexpected errors.
+ */
private void onFailure(final long currentTimeMs,
final Errors responseError) {
handleCoordinatorDisconnect(responseError.exception(), currentTimeMs);
log.debug("Offset fetch failed: {}", responseError.message());
if (responseError == COORDINATOR_LOAD_IN_PROGRESS) {
- retry(currentTimeMs);
+ maybeRetry(currentTimeMs, responseError.exception());
+ } else if (responseError == Errors.UNKNOWN_MEMBER_ID) {
+ log.error("OffsetFetch failed with {} because the member is not part of the group" +
+ " anymore.", responseError);
+ future.completeExceptionally(responseError.exception());
+ } else if (responseError == Errors.STALE_MEMBER_EPOCH) {
+ if (maybeRetryWithNewMemberEpoch(currentTimeMs, responseError)) {
+ log.debug("OffsetFetch failed with {} but the consumer is still part" +
+ " of the group, so the request will be retried with the latest " +
+ "member ID and epoch.", responseError);
+ return;
+ }
+ log.error("OffsetFetch failed with {} and the consumer is not part " +
+ "of the group anymore (it probably left the group, got fenced" +
+ " or failed). The request cannot be retried and will fail.", responseError);
+ future.completeExceptionally(responseError.exception());
} else if (responseError == Errors.NOT_COORDINATOR) {
// re-discover the coordinator and retry
coordinatorRequestManager.markCoordinatorUnknown("error response " + responseError.name(), currentTimeMs);
- retry(currentTimeMs);
+ maybeRetry(currentTimeMs, responseError.exception());
} else if (responseError == Errors.GROUP_AUTHORIZATION_FAILED) {
- future.completeExceptionally(GroupAuthorizationException.forGroupId(groupState.groupId));
+ future.completeExceptionally(GroupAuthorizationException.forGroupId(groupId));
} else {
future.completeExceptionally(new KafkaException("Unexpected error in fetch offset response: " + responseError.message()));
}
}
- private void retry(final long currentTimeMs) {
+ /**
+ * Enqueue the request to be retried with exponential backoff if the time has not expired.
+ * This will fail the request future if the request is not retried.
+ */
+ @Override
+ void maybeRetry(long currentTimeMs, Throwable throwable) {
+ if (isExpired(currentTimeMs)) {
+ future.completeExceptionally(throwable);
+ return;
+ }
onFailedAttempt(currentTimeMs);
+ pendingRequests.inflightOffsetFetches.remove(this);
pendingRequests.addOffsetFetchRequest(this);
}
+ private boolean isExpired(final long currentTimeMs) {
+ return expirationTimeMs <= currentTimeMs;
+ }
+
+ /**
+ * Complete the request future with a TimeoutException if the request expired. No action
+ * taken if the request is still active.
+ *
+ * @return True if the request expired.
+ */
+ private boolean maybeExpire(final long currentTimeMs) {
+ if (isExpired(currentTimeMs)) {
+ future.completeExceptionally(new TimeoutException("OffsetFetch request could not " +
+ "complete before timeout expired."));
+ return true;
+ }
+ return false;
+ }
+
private void onSuccess(final long currentTimeMs,
final OffsetFetchResponse response) {
Set<String> unauthorizedTopics = null;
Map<TopicPartition, OffsetFetchResponse.PartitionData> responseData =
- response.partitionDataMap(groupState.groupId);
+ response.partitionDataMap(groupId);
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(responseData.size());
Set<TopicPartition> unstableTxnOffsetTopicPartitions = new HashSet<>();
for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : responseData.entrySet()) {
@@ -624,13 +869,14 @@ public class CommitRequestManager implements RequestManager {
", this could be either " +
"transactional offsets waiting for completion, or " +
"normal offsets waiting for replication after appending to local log", unstableTxnOffsetTopicPartitions);
- retry(currentTimeMs);
+ maybeRetry(currentTimeMs, Errors.UNSTABLE_OFFSET_COMMIT.exception());
} else {
future.complete(offsets);
}
}
- private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> otherFuture) {
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(
+ final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> otherFuture) {
return this.future.whenComplete((r, t) -> {
if (t != null) {
otherFuture.completeExceptionally(t);
@@ -644,7 +890,8 @@ public class CommitRequestManager implements RequestManager {
public String toString() {
return "OffsetFetchRequestState{" +
"requestedPartitions=" + requestedPartitions +
- ", requestedGeneration=" + requestedGeneration +
+ ", memberId=" + memberInfo.memberId.orElse("undefined") +
+ ", memberEpoch=" + (memberInfo.memberEpoch.isPresent() ? memberInfo.memberEpoch.get() : "undefined") +
", future=" + future +
", " + toStringBase() +
'}';
@@ -671,34 +918,49 @@ public class CommitRequestManager implements RequestManager {
return !unsentOffsetCommits.isEmpty() || !unsentOffsetFetches.isEmpty();
}
- OffsetCommitRequestState addOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ OffsetCommitRequestState addOffsetCommitRequest(
+ final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final Optional<Long> expirationTimeMs,
+ final boolean retryOnStaleEpoch) {
// TODO: Dedupe committing the same offsets to the same partitions
- return addOffsetCommitRequest(createOffsetCommitRequest(offsets, jitter));
+ OffsetCommitRequestState requestState = createOffsetCommitRequest(
+ offsets,
+ jitter,
+ expirationTimeMs,
+ retryOnStaleEpoch);
+ return addOffsetCommitRequest(requestState);
}
OffsetCommitRequestState addOffsetCommitRequest(OffsetCommitRequestState request) {
+ log.debug("Enqueuing OffsetCommit request for offsets: {}", request.offsets);
unsentOffsetCommits.add(request);
return request;
}
OffsetCommitRequestState createOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets,
- final OptionalDouble jitter) {
+ final OptionalDouble jitter,
+ final Optional<Long> expirationTimeMs,
+ final boolean retryOnStaleEpoch) {
return jitter.isPresent() ?
- new OffsetCommitRequestState(
- offsets,
- groupState.groupId,
- groupState.groupInstanceId.orElse(null),
- groupState.generation,
- retryBackoffMs,
- retryBackoffMaxMs,
- jitter.getAsDouble()) :
- new OffsetCommitRequestState(
- offsets,
- groupState.groupId,
- groupState.groupInstanceId.orElse(null),
- groupState.generation,
- retryBackoffMs,
- retryBackoffMaxMs);
+ new OffsetCommitRequestState(
+ offsets,
+ groupId,
+ groupInstanceId,
+ expirationTimeMs,
+ retryBackoffMs,
+ retryBackoffMaxMs,
+ jitter.getAsDouble(),
+ memberInfo,
+ retryOnStaleEpoch) :
+ new OffsetCommitRequestState(
+ offsets,
+ groupId,
+ groupInstanceId,
+ expirationTimeMs,
+ retryBackoffMs,
+ retryBackoffMaxMs,
+ memberInfo,
+ retryOnStaleEpoch);
}
/**
@@ -730,27 +992,30 @@ public class CommitRequestManager implements RequestManager {
return request.future;
}
- private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions) {
+ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> addOffsetFetchRequest(final Set<TopicPartition> partitions,
+ final long expirationTimeMs) {
OffsetFetchRequestState request = jitter.isPresent() ?
new OffsetFetchRequestState(
partitions,
- groupState.generation,
retryBackoffMs,
retryBackoffMaxMs,
- jitter.getAsDouble()) :
+ expirationTimeMs,
+ jitter.getAsDouble(),
+ memberInfo) :
new OffsetFetchRequestState(
partitions,
- groupState.generation,
retryBackoffMs,
- retryBackoffMaxMs);
+ retryBackoffMaxMs,
+ expirationTimeMs,
+ memberInfo);
return addOffsetFetchRequest(request);
}
/**
- * Clear {@code unsentOffsetCommits} and moves all the sendable request in {@code unsentOffsetFetches} to the
- * {@code inflightOffsetFetches} to bookkeep all the inflight requests.
- * Note: Sendable requests are determined by their timer as we are expecting backoff on failed attempt. See
- * {@link RequestState}.
+ * Clear {@code unsentOffsetCommits} and moves all the sendable request in {@code
+ * unsentOffsetFetches} to the {@code inflightOffsetFetches} to bookkeep all the inflight
+ * requests. Note: Sendable requests are determined by their timer as we are expecting
+ * backoff on failed attempt. See {@link RequestState}.
*/
List<NetworkClientDelegate.UnsentRequest> drain(final long currentTimeMs) {
List<NetworkClientDelegate.UnsentRequest> unsentRequests = new ArrayList<>();
@@ -760,6 +1025,8 @@ public class CommitRequestManager implements RequestManager {
.filter(request -> !request.canSendRequest(currentTimeMs))
.collect(Collectors.toList());
+ failAndRemoveExpiredCommitRequests(currentTimeMs);
+
// Add all unsent offset commit requests to the unsentRequests list
unsentRequests.addAll(
unsentOffsetCommits.stream()
@@ -773,6 +1040,8 @@ public class CommitRequestManager implements RequestManager {
unsentOffsetFetches.stream()
.collect(Collectors.partitioningBy(request -> request.canSendRequest(currentTimeMs)));
+ failAndRemoveExpiredFetchRequests(currentTimeMs);
+
// Add all sendable offset fetch requests to the unsentRequests list and to the inflightOffsetFetches list
for (OffsetFetchRequestState request : partitionedBySendability.get(true)) {
request.onSendAttempt(currentTimeMs);
@@ -788,6 +1057,22 @@ public class CommitRequestManager implements RequestManager {
return Collections.unmodifiableList(unsentRequests);
}
+ /**
+ * Find the unsent commit requests that have expired, remove them and complete their
+ * futures with a TimeoutException.
+ */
+ private void failAndRemoveExpiredCommitRequests(final long currentTimeMs) {
+ unsentOffsetCommits.removeIf(req -> req.maybeExpire(currentTimeMs));
+ }
+
+ /**
+ * Find the unsent fetch requests that have expired, remove them and complete their
+ * futures with a TimeoutException.
+ */
+ private void failAndRemoveExpiredFetchRequests(final long currentTimeMs) {
+ unsentOffsetFetches.removeIf(req -> req.maybeExpire(currentTimeMs));
+ }
+
private void clearAll() {
unsentOffsetCommits.clear();
unsentOffsetFetches.clear();
@@ -825,6 +1110,10 @@ public class CommitRequestManager implements RequestManager {
this.timer.reset(autoCommitInterval);
}
+ public void resetTimer(long retryBackoffMs) {
+ this.timer.reset(retryBackoffMs);
+ }
+
public long remainingMs(final long currentTimeMs) {
this.timer.update(currentTimeMs);
return this.timer.remainingMs();
@@ -838,4 +1127,14 @@ public class CommitRequestManager implements RequestManager {
this.hasInflightCommit = inflightCommitStatus;
}
}
+
+ static class MemberInfo {
+ Optional<String> memberId;
+ Optional<Integer> memberEpoch;
+
+ MemberInfo() {
+ this.memberId = Optional.empty();
+ this.memberEpoch = Optional.empty();
+ }
+ }
}
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
deleted file mode 100644
index 4e16055f6fd..00000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/GroupState.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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<String> groupInstanceId;
- public Generation generation = Generation.NO_GENERATION;
-
- public GroupState(String groupId, Optional<String> 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/MemberStateListener.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberStateListener.java
new file mode 100644
index 00000000000..a4084d78931
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberStateListener.java
@@ -0,0 +1,37 @@
+/*
+ * 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 java.util.Optional;
+
+/**
+ * Listener for getting notified of member ID and epoch changes.
+ */
+public interface MemberStateListener {
+
+ /**
+ * Called whenever member ID or epoch change with new values received from the broker or
+ * cleared if the member is not part of the group anymore (when it gets fenced, leaves the
+ * group or fails).
+ *
+ * @param memberEpoch New member epoch received from the broker. Empty if the member is
+ * not part of the group anymore.
+ * @param memberId Current member ID. Empty if the member is not part of the group.
+ */
+ void onMemberEpochUpdated(Optional<Integer> memberEpoch, Optional<String> memberId);
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
index aec25985093..40504966223 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
@@ -168,4 +168,10 @@ public interface MembershipManager {
* the member to STALE.
*/
void transitionToStaled();
+
+ /**
+ * Register a listener that will be called whenever the member state changes due to
+ * transitions of new data received from the server, as defined in {@link MemberStateListener}.
+ */
+ void registerStateListener(MemberStateListener listener);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
index 2de533eaa2a..b18a01b7240 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
@@ -41,10 +41,12 @@ import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -114,15 +116,13 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
/**
* TopicPartition comparator based on topic name and partition id.
*/
- final static TopicPartitionComparator TOPIC_PARTITION_COMPARATOR =
- new TopicPartitionComparator();
+ final static TopicPartitionComparator TOPIC_PARTITION_COMPARATOR = new TopicPartitionComparator();
/**
* TopicIdPartition comparator based on topic name and partition id (ignoring ID while sorting,
* as this is sorted mainly for logging purposes).
*/
- final static TopicIdPartitionComparator TOPIC_ID_PARTITION_COMPARATOR =
- new TopicIdPartitionComparator();
+ final static TopicIdPartitionComparator TOPIC_ID_PARTITION_COMPARATOR = new TopicIdPartitionComparator();
/**
* Group ID of the consumer group the member will be part of, provided when creating the current
@@ -135,6 +135,13 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
*/
private final Optional<String> groupInstanceId;
+ /**
+ * Rebalance timeout. To be used as time limit for the commit request issued
+ * when a new assignment is received, that is retried until it succeeds, fails with a
+ * non-retriable error, it the time limit expires.
+ */
+ private final int rebalanceTimeoutMs;
+
/**
* Member ID assigned by the server to the member, received in a heartbeat response when
* joining the group specified in {@link #groupId}
@@ -247,6 +254,13 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
*/
private boolean isRegisteredForMetadataUpdates;
+ /**
+ * Registered listeners that will be notified whenever the memberID/epoch gets updated (valid
+ * values received from the broker, or values cleared due to member leaving the group, getting
+ * fenced or failing).
+ */
+ private final List<MemberStateListener> stateUpdatesListeners;
+
/**
* Optional client telemetry reporter which sends client telemetry data to the broker. This
* will be empty if the client telemetry feature is not enabled. This is provided to update
@@ -263,6 +277,7 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
public MembershipManagerImpl(String groupId,
Optional<String> groupInstanceId,
+ int rebalanceTimeoutMs,
Optional<String> serverAssignor,
SubscriptionState subscriptions,
CommitRequestManager commitRequestManager,
@@ -282,7 +297,9 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
this.assignmentReadyToReconcile = new TreeSet<>(TOPIC_ID_PARTITION_COMPARATOR);
this.currentAssignment = new HashMap<>();
this.log = logContext.logger(MembershipManagerImpl.class);
+ this.stateUpdatesListeners = new ArrayList<>();
this.clientTelemetryReporter = clientTelemetryReporter;
+ this.rebalanceTimeoutMs = rebalanceTimeoutMs;
this.backgroundEventHandler = backgroundEventHandler;
}
@@ -361,8 +378,8 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
}
this.memberId = response.memberId();
- this.memberEpoch = response.memberEpoch();
- commitRequestManager.updateMemberInformation(this.memberId, this.memberEpoch);
+ updateMemberEpoch(response.memberEpoch());
+
ConsumerGroupHeartbeatResponseData.Assignment assignment = response.assignment();
if (assignment != null) {
@@ -483,6 +500,7 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
MemberState previousState = state;
transitionTo(MemberState.FATAL);
log.error("Member {} with epoch {} transitioned to {} state", memberId, memberEpoch, MemberState.FATAL);
+ notifyEpochChange(Optional.empty(), Optional.empty());
if (previousState == MemberState.UNSUBSCRIBED) {
log.debug("Member {} with epoch {} got fatal error from the broker but it already " +
@@ -646,14 +664,23 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
memberId);
return;
}
- memberEpoch = groupInstanceId.isPresent() ?
+ int leaveEpoch = groupInstanceId.isPresent() ?
ConsumerGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH :
ConsumerGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH;
- commitRequestManager.updateMemberInformation(this.memberId, this.memberEpoch);
+ updateMemberEpoch(leaveEpoch);
currentAssignment = new HashMap<>();
transitionTo(MemberState.LEAVING);
}
+ /**
+ * Call all listeners that are registered to get notified when the member epoch is updated.
+ * This also includes the latest member ID in the notification. If the member fails or leaves
+ * the group, this will be invoked with empty epoch and member ID.
+ */
+ private void notifyEpochChange(Optional<Integer> epoch, Optional<String> memberId) {
+ stateUpdatesListeners.forEach(stateListener -> stateListener.onMemberEpochUpdated(epoch, memberId));
+ }
+
/**
* {@inheritDoc}
*/
@@ -794,36 +821,76 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
revokedPartitions
);
+ // Commit offsets if auto-commit enabled before reconciling a new assignment. Request will
+ // be retried until it succeeds, fails with non-retriable error, or timer expires.
+ CompletableFuture<Void> commitResult;
+
+ // Issue a commit request that will be retried until it succeeds, fails with a
+ // non-retriable error, or the time limit expires. Retry on stale member epoch error, in a
+ // best effort to commit the offsets in the case where the epoch might have changed while
+ // the current reconciliation is in process. Note this is using the rebalance timeout as
+ // it is the limit enforced by the broker to complete the reconciliation process.
+ commitResult = commitRequestManager.maybeAutoCommitAllConsumedNow(
+ Optional.of((long) rebalanceTimeoutMs),
+ true);
+
+ // Execute commit -> onPartitionsRevoked -> onPartitionsAssigned.
+ commitResult.whenComplete((commitReqResult, commitReqError) -> {
+ if (commitReqError != null) {
+ // The call to commit, that includes retry logic for retriable errors, failed to
+ // complete within the time boundaries (fatal error or retriable that did not
+ // recover). Proceed with the revocation.
+ log.error("Auto-commit request before reconciling new assignment failed. " +
+ "Will proceed with the reconciliation anyway.", commitReqError);
+ } else {
+ log.debug("Auto-commit before reconciling new assignment completed successfully.");
+ }
+
+ revokeAndAssign(assignedTopicIdPartitions, revokedPartitions, addedPartitions);
+ });
+
+ return true;
+ }
+
+ /**
+ * Trigger onPartitionsRevoked callbacks if any partitions where revoked. If it succeeds,
+ * proceed to trigger the onPartitionsAssigned (even if no new partitions were added), and
+ * then complete the reconciliation by updating the assignment and making the appropriate state
+ * transition. Note that if any of the 2 callbacks fails, the reconciliation should fail.
+ */
+ private void revokeAndAssign(SortedSet<TopicIdPartition> assignedTopicIdPartitions,
+ SortedSet<TopicPartition> revokedPartitions,
+ SortedSet<TopicPartition> addedPartitions) {
CompletableFuture<Void> revocationResult;
if (!revokedPartitions.isEmpty()) {
revocationResult = revokePartitions(revokedPartitions);
} else {
revocationResult = CompletableFuture.completedFuture(null);
- // Reschedule the auto commit starting from now (new assignment received without any
- // revocation).
- commitRequestManager.resetAutoCommitTimer();
}
// Future that will complete when the full reconciliation process completes (revocation
// and assignment, executed sequentially).
CompletableFuture<Void> reconciliationResult =
- revocationResult.thenCompose(__ -> {
- boolean memberHasRejoined = memberEpochOnReconciliationStart != memberEpoch;
- if (state == MemberState.RECONCILING && !memberHasRejoined) {
- // Apply assignment
- return assignPartitions(assignedTopicIdPartitions, addedPartitions);
- } else {
- log.debug("Revocation callback completed but the member already " +
- "transitioned out of the reconciling state for epoch {} into " +
- "{} state with epoch {}. Interrupting reconciliation as it's " +
- "not relevant anymore,", memberEpochOnReconciliationStart, state, memberEpoch);
- String reason = interruptedReconciliationErrorMessage();
- CompletableFuture<Void> res = new CompletableFuture<>();
- res.completeExceptionally(new KafkaException("Interrupting reconciliation" +
- " after revocation. " + reason));
- return res;
- }
- });
+ revocationResult.thenCompose(__ -> {
+ boolean memberHasRejoined = memberEpochOnReconciliationStart != memberEpoch;
+ if (state == MemberState.RECONCILING && !memberHasRejoined) {
+ // Reschedule the auto commit starting from now that member has a new assignment.
+ commitRequestManager.resetAutoCommitTimer();
+
+ // Apply assignment
+ return assignPartitions(assignedTopicIdPartitions, addedPartitions);
+ } else {
+ log.debug("Revocation callback completed but the member already " +
+ "transitioned out of the reconciling state for epoch {} into " +
+ "{} state with epoch {}. Interrupting reconciliation as it's " +
+ "not relevant anymore,", memberEpochOnReconciliationStart, state, memberEpoch);
+ String reason = interruptedReconciliationErrorMessage();
+ CompletableFuture<Void> res = new CompletableFuture<>();
+ res.completeExceptionally(new KafkaException("Interrupting reconciliation" +
+ " after revocation. " + reason));
+ return res;
+ }
+ });
reconciliationResult.whenComplete((result, error) -> {
markReconciliationCompleted();
@@ -846,12 +913,10 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
} else {
String reason = interruptedReconciliationErrorMessage();
log.error("Interrupting reconciliation after partitions assigned callback " +
- "completed. " + reason);
+ "completed. " + reason);
}
}
});
-
- return true;
}
// Visible for testing.
@@ -1007,47 +1072,29 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
// request and user callback execution).
CompletableFuture<Void> revocationResult = new CompletableFuture<>();
- // Commit offsets if auto-commit enabled.
- CompletableFuture<Void> commitResult;
- if (commitRequestManager.autoCommitEnabled()) {
- commitResult = commitRequestManager.maybeAutoCommitAllConsumed();
- } else {
- commitResult = CompletableFuture.completedFuture(null);
+ // At this point we expect to be in a middle of a revocation triggered from RECONCILING
+ // or PREPARE_LEAVING, but it could be the case that the member received a fatal error
+ // while waiting for the commit to complete. Check if that's the case and abort the
+ // revocation.
+ if (state == MemberState.FATAL) {
+ String errorMsg = String.format("Member %s with epoch %s received a fatal error " +
+ "while waiting for a revocation commit to complete. Will abort revocation " +
+ "without triggering user callback.", memberId, memberEpoch);
+ log.debug(errorMsg);
+ revocationResult.completeExceptionally(new KafkaException(errorMsg));
+ return revocationResult;
}
- commitResult.whenComplete((result, error) -> {
- if (error != null) {
- // Commit request failed (commit request manager internally retries on
- // retriable errors, so at this point we assume this is non-retriable, but
- // proceed with the revocation anyway).
- log.error("Commit request before revocation failed with non-retriable error. Will" +
- " proceed with the revocation anyway.", error);
- }
-
- // At this point we expect to be in a middle of a revocation triggered from RECONCILING
- // or PREPARE_LEAVING, but it could be the case that the member received a fatal error
- // while waiting for the commit to complete. Check if that's the case and abort the
- // revocation.
- if (state == MemberState.FATAL) {
- String errorMsg = String.format("Member %s with epoch %s received a fatal error " +
- "while waiting for a revocation commit to complete. Will abort revocation " +
- "without triggering user callback.", memberId, memberEpoch);
- log.debug(errorMsg);
- revocationResult.completeExceptionally(new KafkaException(errorMsg));
- return;
+ CompletableFuture<Void> userCallbackResult = invokeOnPartitionsRevokedCallback(revokedPartitions);
+ userCallbackResult.whenComplete((callbackResult, callbackError) -> {
+ if (callbackError != null) {
+ log.error("onPartitionsRevoked callback invocation failed for partitions {}",
+ revokedPartitions, callbackError);
+ revocationResult.completeExceptionally(callbackError);
+ } else {
+ revocationResult.complete(null);
}
- CompletableFuture<Void> userCallbackResult = invokeOnPartitionsRevokedCallback(revokedPartitions);
- userCallbackResult.whenComplete((callbackResult, callbackError) -> {
- if (callbackError != null) {
- log.error("onPartitionsRevoked callback invocation failed for partitions {}",
- revokedPartitions, callbackError);
- revocationResult.completeExceptionally(callbackError);
- } else {
- revocationResult.complete(null);
- }
-
- });
});
return revocationResult;
}
@@ -1208,8 +1255,21 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
}
private void resetEpoch() {
- this.memberEpoch = ConsumerGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH;
- commitRequestManager.updateMemberInformation(this.memberId, this.memberEpoch);
+ updateMemberEpoch(ConsumerGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH);
+ }
+
+ private void updateMemberEpoch(int newEpoch) {
+ boolean newEpochReceived = this.memberEpoch != newEpoch;
+ this.memberEpoch = newEpoch;
+ // Simply notify based on epoch change only, given that the member will never receive a
+ // new member ID without an epoch (member ID is only assigned when it joins the group).
+ if (newEpochReceived) {
+ if (memberEpoch > 0) {
+ notifyEpochChange(Optional.of(memberEpoch), Optional.ofNullable(memberId));
+ } else {
+ notifyEpochChange(Optional.empty(), Optional.empty());
+ }
+ }
}
/**
@@ -1278,4 +1338,18 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
reconcile();
}
}
+
+ /**
+ * Register a new listener that will be invoked whenever the member state changes, or a new
+ * member ID or epoch is received.
+ *
+ * @param listener Listener to invoke.
+ */
+ @Override
+ public void registerStateListener(MemberStateListener listener) {
+ if (listener == null) {
+ throw new IllegalArgumentException("State updates listener cannot be null");
+ }
+ this.stateUpdatesListeners.add(listener);
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
index 935402b884d..5305c004b95 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
@@ -154,23 +154,24 @@ public class RequestManagers implements Closeable {
if (groupRebalanceConfig != null && groupRebalanceConfig.groupId != null) {
Optional<String> serverAssignor = Optional.ofNullable(config.getString(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG));
- final GroupState groupState = new GroupState(groupRebalanceConfig);
coordinator = new CoordinatorRequestManager(time,
logContext,
retryBackoffMs,
retryBackoffMaxMs,
backgroundEventHandler,
- groupState.groupId);
- commit = new CommitRequestManager(time,
+ groupRebalanceConfig.groupId);
+ commit = new CommitRequestManager(
+ time,
logContext,
subscriptions,
config,
coordinator,
- backgroundEventHandler,
- groupState);
+ groupRebalanceConfig.groupId,
+ groupRebalanceConfig.groupInstanceId);
membershipManager = new MembershipManagerImpl(
- groupState.groupId,
- groupState.groupInstanceId,
+ groupRebalanceConfig.groupId,
+ groupRebalanceConfig.groupInstanceId,
+ groupRebalanceConfig.rebalanceTimeoutMs,
serverAssignor,
subscriptions,
commit,
@@ -178,6 +179,7 @@ public class RequestManagers implements Closeable {
logContext,
clientTelemetryReporter,
backgroundEventHandler);
+ membershipManager.registerStateListener(commit);
heartbeatRequestManager = new HeartbeatRequestManager(
logContext,
time,
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 8359cb8320c..d3aae9ae1a8 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
@@ -33,6 +33,7 @@ import org.slf4j.Logger;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
@@ -149,7 +150,8 @@ public class ApplicationEventProcessor extends EventProcessor<ApplicationEvent>
}
CommitRequestManager manager = requestManagers.commitRequestManager.get();
- event.chain(manager.addOffsetCommitRequest(event.offsets()));
+ Optional<Long> expirationTimeMs = event.retryTimeoutMs().map(this::getExpirationTimeForTimeout);
+ event.chain(manager.addOffsetCommitRequest(event.offsets(), expirationTimeMs, false));
}
private void process(final FetchCommittedOffsetsApplicationEvent event) {
@@ -159,20 +161,26 @@ public class ApplicationEventProcessor extends EventProcessor<ApplicationEvent>
return;
}
CommitRequestManager manager = requestManagers.commitRequestManager.get();
- event.chain(manager.addOffsetFetchRequest(event.partitions()));
+ long expirationTimeMs = getExpirationTimeForTimeout(event.timeout());
+ event.chain(manager.addOffsetFetchRequest(event.partitions(), expirationTimeMs));
}
private void process(final NewTopicsMetadataUpdateRequestEvent ignored) {
metadata.requestUpdateForNewTopics();
}
+
+ /**
+ * Commit all consumed if auto-commit is enabled. Note this will trigger an async commit,
+ * that will not be retried if the commit request fails.
+ */
private void process(final AssignmentChangeApplicationEvent event) {
if (!requestManagers.commitRequestManager.isPresent()) {
return;
}
CommitRequestManager manager = requestManagers.commitRequestManager.get();
manager.updateAutoCommitTimer(event.currentTimeMs());
- manager.maybeAutoCommit(event.offsets());
+ manager.maybeAutoCommitAllConsumedAsync();
}
private void process(final ListOffsetsApplicationEvent event) {
@@ -228,8 +236,7 @@ public class ApplicationEventProcessor extends EventProcessor<ApplicationEvent>
private void process(final TopicMetadataApplicationEvent event) {
final CompletableFuture<Map<String, List<PartitionInfo>>> future;
- long expirationTimeMs =
- (event.getTimeoutMs() == Long.MAX_VALUE) ? Long.MAX_VALUE : System.currentTimeMillis() + event.getTimeoutMs();
+ long expirationTimeMs = getExpirationTimeForTimeout(event.getTimeoutMs());
if (event.isAllTopics()) {
future = requestManagers.topicMetadataRequestManager.requestAllTopicsMetadata(expirationTimeMs);
} else {
@@ -272,6 +279,19 @@ public class ApplicationEventProcessor extends EventProcessor<ApplicationEvent>
event.chain(future);
}
+ /**
+ * @return Expiration time in milliseconds calculated with the current time plus the given
+ * timeout. Returns Long.MAX_VALUE if the expiration overflows it.
+ * Visible for testing.
+ */
+ long getExpirationTimeForTimeout(final long timeoutMs) {
+ long expiration = System.currentTimeMillis() + timeoutMs;
+ if (expiration < 0) {
+ return Long.MAX_VALUE;
+ }
+ return expiration;
+ }
+
/**
* Creates a {@link Supplier} for deferred creation during invocation by
* {@link ConsumerNetworkThread}.
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 48e62fe78bf..d2205227c49 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
@@ -21,14 +21,32 @@ import org.apache.kafka.common.TopicPartition;
import java.util.Collections;
import java.util.Map;
+import java.util.Optional;
public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
+ /**
+ * Offsets to commit per partition.
+ */
private final Map<TopicPartition, OffsetAndMetadata> offsets;
- public CommitApplicationEvent(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+ /**
+ * Time to wait for a response, retrying on retriable errors. If not present, the request is
+ * triggered without waiting for a response or being retried.
+ */
+ private final Optional<Long> retryTimeoutMs;
+
+ /**
+ * Create new event to commit offsets. If timer is present, the request will be retried on
+ * retriable errors until the timer expires (sync commit offsets request). If the timer is
+ * not present, the request will be sent without waiting for a response of retrying (async
+ * commit offsets request).
+ */
+ public CommitApplicationEvent(final Map<TopicPartition, OffsetAndMetadata> offsets,
+ final Optional<Long> retryTimeoutMs) {
super(Type.COMMIT);
this.offsets = Collections.unmodifiableMap(offsets);
+ this.retryTimeoutMs = retryTimeoutMs;
for (OffsetAndMetadata offsetAndMetadata : offsets.values()) {
if (offsetAndMetadata.offset() < 0) {
@@ -41,6 +59,10 @@ public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
return offsets;
}
+ public Optional<Long> retryTimeoutMs() {
+ return retryTimeoutMs;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
@@ -64,6 +86,7 @@ public class CommitApplicationEvent extends CompletableApplicationEvent<Void> {
return "CommitApplicationEvent{" +
toStringBase() +
", offsets=" + offsets +
+ ", retryTimeout=" + (retryTimeoutMs.map(t -> t + "ms").orElse("none")) +
'}';
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/FetchCommittedOffsetsApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/FetchCommittedOffsetsApplicationEvent.java
index 6765a408b60..34b2d97705c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/FetchCommittedOffsetsApplicationEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/FetchCommittedOffsetsApplicationEvent.java
@@ -25,17 +25,31 @@ import java.util.Set;
public class FetchCommittedOffsetsApplicationEvent extends CompletableApplicationEvent<Map<TopicPartition, OffsetAndMetadata>> {
+ /**
+ * Partitions to retrieve committed offsets for.
+ */
private final Set<TopicPartition> partitions;
- public FetchCommittedOffsetsApplicationEvent(final Set<TopicPartition> partitions) {
+ /**
+ * Time until which the request will be retried if it fails with a retriable error.
+ */
+ private final long timeoutMs;
+
+ public FetchCommittedOffsetsApplicationEvent(final Set<TopicPartition> partitions,
+ final long timeoutMs) {
super(Type.FETCH_COMMITTED_OFFSETS);
this.partitions = Collections.unmodifiableSet(partitions);
+ this.timeoutMs = timeoutMs;
}
public Set<TopicPartition> partitions() {
return partitions;
}
+ public long timeout() {
+ return timeoutMs;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
@@ -59,6 +73,7 @@ public class FetchCommittedOffsetsApplicationEvent extends CompletableApplicatio
return getClass().getSimpleName() + "{" +
toStringBase() +
", partitions=" + partitions +
+ ", timeout=" + timeoutMs + "ms" +
'}';
}
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index ac052d72f6f..05903d44120 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
@@ -54,7 +54,6 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.InvalidGroupIdException;
-import org.apache.kafka.common.errors.NetworkException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.WakeupException;
@@ -447,18 +446,13 @@ public class AsyncKafkaConsumerTest {
@Test
public void testEnsureCommitSyncExecutedCommitAsyncCallbacks() {
consumer = newConsumer();
- final HashMap<TopicPartition, OffsetAndMetadata> offsets = mockTopicPartitionOffset();
- MockCommitCallback callback = new MockCommitCallback();
- assertDoesNotThrow(() -> consumer.commitAsync(offsets, callback));
+ KafkaException callbackException = new KafkaException("Async commit callback failed");
+ OffsetCommitCallback callback = (offsets, exception) -> {
+ throw callbackException;
+ };
- final ArgumentCaptor<CommitApplicationEvent> commitEventCaptor = ArgumentCaptor.forClass(CommitApplicationEvent.class);
- verify(applicationEventHandler).add(commitEventCaptor.capture());
- final CommitApplicationEvent commitEvent = commitEventCaptor.getValue();
- commitEvent.future().completeExceptionally(new NetworkException("Test exception"));
-
- assertMockCommitCallbackInvoked(() -> consumer.commitSync(),
- callback,
- Errors.NETWORK_EXCEPTION);
+ assertDoesNotThrow(() -> consumer.commitAsync(new HashMap<>(), callback));
+ assertThrows(callbackException.getClass(), () -> consumer.commitSync());
}
@Test
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 f8cecba8f99..22d2be1a820 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
@@ -17,16 +17,19 @@
package org.apache.kafka.clients.consumer.internals;
import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
-import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
+import org.apache.kafka.clients.consumer.RetriableCommitFailedException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.DisconnectException;
+import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.message.OffsetCommitRequestData;
import org.apache.kafka.common.message.OffsetCommitResponseData;
+import org.apache.kafka.common.message.OffsetFetchRequestData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractRequest;
@@ -53,10 +56,11 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.OptionalDouble;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -67,8 +71,11 @@ import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
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.apache.kafka.clients.consumer.internals.ConsumerTestBuilder.DEFAULT_GROUP_ID;
+import static org.apache.kafka.clients.consumer.internals.ConsumerTestBuilder.DEFAULT_GROUP_INSTANCE_ID;
+import static org.apache.kafka.test.TestUtils.assertFutureThrows;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -83,12 +90,13 @@ public class CommitRequestManagerTest {
private long retryBackoffMaxMs = 1000;
private Node mockedNode = new Node(1, "host1", 9092);
private SubscriptionState subscriptionState;
- private GroupState groupState;
private LogContext logContext;
private MockTime time;
private CoordinatorRequestManager coordinatorRequestManager;
private Properties props;
- private BackgroundEventHandler backgroundEventHandler;
+
+ private final int defaultApiTimeoutMs = 60000;
+
@BeforeEach
public void setup() {
@@ -96,8 +104,6 @@ public class CommitRequestManagerTest {
this.time = new MockTime(0);
this.subscriptionState = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST);
this.coordinatorRequestManager = mock(CoordinatorRequestManager.class);
- this.groupState = new GroupState(DEFAULT_GROUP_ID, Optional.empty());
- this.backgroundEventHandler = spy(new BackgroundEventHandler(logContext, new LinkedBlockingQueue<>()));
this.props = new Properties();
this.props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
this.props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
@@ -105,37 +111,35 @@ public class CommitRequestManagerTest {
}
@Test
- public void testPoll_SkipIfCoordinatorUnknown() {
+ public void testPollSkipIfCoordinatorUnknown() {
CommitRequestManager commitRequestManger = create(false, 0);
assertPoll(false, 0, commitRequestManger);
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0));
- commitRequestManger.addOffsetCommitRequest(offsets);
+ commitRequestManger.addOffsetCommitRequest(offsets, Optional.empty(), false);
assertPoll(false, 0, commitRequestManger);
}
@Test
- public void testPoll_EnsureManualCommitSent() {
+ public void testPollEnsureManualCommitSent() {
CommitRequestManager commitRequestManger = create(false, 0);
assertPoll(0, commitRequestManger);
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0));
- commitRequestManger.addOffsetCommitRequest(offsets);
+ commitRequestManger.addOffsetCommitRequest(offsets, Optional.empty(), false);
assertPoll(1, commitRequestManger);
}
@Test
- public void testPoll_EnsureAutocommitSent() {
+ public void testPollEnsureAutocommitSent() {
TopicPartition tp = new TopicPartition("t1", 1);
subscriptionState.assignFromUser(Collections.singleton(tp));
subscriptionState.seek(tp, 100);
CommitRequestManager commitRequestManger = create(true, 100);
assertPoll(0, commitRequestManger);
- Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
- offsets.put(tp, new OffsetAndMetadata(0));
commitRequestManger.updateAutoCommitTimer(time.milliseconds());
time.sleep(100);
commitRequestManger.updateAutoCommitTimer(time.milliseconds());
@@ -148,7 +152,7 @@ public class CommitRequestManagerTest {
}
@Test
- public void testPoll_EnsureCorrectInflightRequestBufferSize() {
+ public void testPollEnsureCorrectInflightRequestBufferSize() {
CommitRequestManager commitManager = create(false, 100);
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
@@ -163,10 +167,11 @@ public class CommitRequestManagerTest {
// Add the requests to the CommitRequestManager and store their futures
ArrayList<CompletableFuture<Void>> commitFutures = new ArrayList<>();
ArrayList<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> fetchFutures = new ArrayList<>();
- commitFutures.add(commitManager.addOffsetCommitRequest(offsets1));
- fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 0))));
- commitFutures.add(commitManager.addOffsetCommitRequest(offsets2));
- fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 1))));
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
+ commitFutures.add(commitManager.addOffsetCommitRequest(offsets1, Optional.of(expirationTimeMs), false));
+ fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 0)), expirationTimeMs));
+ commitFutures.add(commitManager.addOffsetCommitRequest(offsets2, Optional.of(expirationTimeMs), false));
+ fetchFutures.add(commitManager.addOffsetFetchRequest(Collections.singleton(new TopicPartition("test", 1)), expirationTimeMs));
// Poll the CommitRequestManager and verify that the inflightOffsetFetches size is correct
NetworkClientDelegate.PollResult result = commitManager.poll(time.milliseconds());
@@ -185,41 +190,50 @@ public class CommitRequestManagerTest {
}
@Test
- public void testPoll_EnsureEmptyPendingRequestAfterPoll() {
+ public void testPollEnsureEmptyPendingRequestAfterPoll() {
CommitRequestManager commitRequestManger = create(true, 100);
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
- commitRequestManger.addOffsetCommitRequest(new HashMap<>());
+ Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(
+ new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+ commitRequestManger.addOffsetCommitRequest(offsets, Optional.empty(), false);
assertEquals(1, commitRequestManger.unsentOffsetCommitRequests().size());
assertEquals(1, commitRequestManger.poll(time.milliseconds()).unsentRequests.size());
assertTrue(commitRequestManger.unsentOffsetCommitRequests().isEmpty());
assertEmptyPendingRequests(commitRequestManger);
}
+ // This is the case of the async auto commit sent on calls to assign (async commit that
+ // should not be retried).
@Test
- public void testAutocommit_ResendAutocommitAfterException() {
- CommitRequestManager commitRequestManger = create(true, 100);
+ public void testAsyncAutocommitNotRetriedAfterException() {
+ long commitInterval = retryBackoffMs * 2;
+ CommitRequestManager commitRequestManger = create(true, commitInterval);
TopicPartition tp = new TopicPartition("topic", 1);
subscriptionState.assignFromUser(Collections.singleton(tp));
subscriptionState.seek(tp, 100);
- time.sleep(100);
+ time.sleep(commitInterval);
commitRequestManger.updateAutoCommitTimer(time.milliseconds());
List<NetworkClientDelegate.FutureCompletionHandler> futures = assertPoll(1, commitRequestManger);
- time.sleep(99);
- // complete the autocommit request (exceptionally), and expect an exponential backoff of 100ms
+ // Complete the autocommit request exceptionally. It should fail right away, without retry.
futures.get(0).onComplete(mockOffsetCommitResponse(
"topic",
1,
(short) 1,
Errors.COORDINATOR_LOAD_IN_PROGRESS));
- // Expecting to wait for 100ms but only waited 99ms. No result is expected here.
- time.sleep(99);
+ // When polling again before the auto-commit interval no request should be generated
+ // (making sure we wait for the backoff, to check that the failed request is not being
+ // retried).
+ time.sleep(retryBackoffMs);
commitRequestManger.updateAutoCommitTimer(time.milliseconds());
assertPoll(0, commitRequestManger);
commitRequestManger.updateAutoCommitTimer(time.milliseconds());
- // Complete the full backoff of 100ms, and now expecting a result.
- time.sleep(1);
+ // Only when polling after the auto-commit interval, a new auto-commit request should be
+ // generated.
+ time.sleep(commitInterval);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
futures = assertPoll(1, commitRequestManger);
assertEmptyPendingRequests(commitRequestManger);
futures.get(0).onComplete(mockOffsetCommitResponse(
@@ -229,33 +243,132 @@ public class CommitRequestManagerTest {
Errors.NONE));
}
+ // This is the case of the sync auto commit sent when the consumer is being closed (sync commit
+ // that should be retried until it succeeds, fails, or timer expires).
+ @ParameterizedTest
+ @MethodSource("offsetCommitExceptionSupplier")
+ public void testSyncAutocommitRetriedAfterRetriableException(Errors error) {
+ long commitInterval = retryBackoffMs * 2;
+ CommitRequestManager commitRequestManger = create(true, commitInterval);
+ TopicPartition tp = new TopicPartition("topic", 1);
+ subscriptionState.assignFromUser(Collections.singleton(tp));
+ subscriptionState.seek(tp, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ time.sleep(commitInterval);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+
+ // Auto-commit all consume sync (ex. triggered when the consumer is closed).
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
+ CompletableFuture<Void> commitResult =
+ commitRequestManger.maybeAutoCommitAllConsumedNow(Optional.of(expirationTimeMs), false);
+ sendAndVerifyOffsetCommitRequestFailedAndMaybeRetried(commitRequestManger, error, commitResult);
+
+ // We expect that request should have been retried on this sync commit.
+ assertExceptionHandling(commitRequestManger, error, true);
+ assertCoordinatorDisconnect(error);
+ }
+
+ @Test
+ public void testOffsetCommitFailsWithCommitFailedExceptionIfUnknownMemberId() {
+ long commitInterval = retryBackoffMs * 2;
+ CommitRequestManager commitRequestManger = create(true, commitInterval);
+ TopicPartition tp = new TopicPartition("topic", 1);
+ subscriptionState.assignFromUser(Collections.singleton(tp));
+ subscriptionState.seek(tp, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+ time.sleep(commitInterval);
+ commitRequestManger.updateAutoCommitTimer(time.milliseconds());
+
+ // Auto-commit all consume sync (ex. triggered when the consumer is closed).
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
+ CompletableFuture<Void> commitResult =
+ commitRequestManger.maybeAutoCommitAllConsumedNow(Optional.of(expirationTimeMs), false);
+
+ completeOffsetCommitRequestWithError(commitRequestManger, Errors.UNKNOWN_MEMBER_ID);
+ NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ // Commit should fail with CommitFailedException
+ assertTrue(commitResult.isDone());
+ assertFutureThrows(commitResult, CommitFailedException.class);
+ }
+
+ /**
+ * This is the case where a request to commit offsets is performed without retrying on
+ * STALE_MEMBER_EPOCH (ex. commits triggered from the consumer API, auto-commits triggered on
+ * the interval). The expectation is that the request should fail with CommitFailedException.
+ */
+ @Test
+ public void testOffsetCommitFailsWithCommitFailedExceptionIfStaleMemberEpochNotRetried() {
+ CommitRequestManager commitRequestManger = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(
+ new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+
+ // Send commit request expected to be retried on retriable errors
+ CompletableFuture<Void> commitResult = commitRequestManger.addOffsetCommitRequest(
+ offsets, Optional.of(time.milliseconds() + defaultApiTimeoutMs), false);
+ completeOffsetCommitRequestWithError(commitRequestManger, Errors.STALE_MEMBER_EPOCH);
+ NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+
+ // Commit should fail with CommitFailedException
+ assertTrue(commitResult.isDone());
+ assertFutureThrows(commitResult, CommitFailedException.class);
+ }
+
+ /**
+ * This is the case of the async auto commit request triggered on the interval. The
+ * expectation is that the request should fail with RetriableCommitFailedException, so that
+ * the timer is reset and the request is retried on the next interval.
+ */
@Test
- public void testAutocommit_EnsureOnlyOneInflightRequest() {
+ public void testOffsetAutoCommitOnIntervalFailsWithRetriableOnStaleMemberEpoch() {
+ CommitRequestManager commitRequestManger = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(
+ new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+
+ // Async auto-commit that won't be retried.
+ CompletableFuture<Void> commitResult = commitRequestManger.addOffsetCommitRequest(
+ offsets, Optional.empty(), true);
+ completeOffsetCommitRequestWithError(commitRequestManger, Errors.STALE_MEMBER_EPOCH);
+ NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+
+ // Commit should fail with RetriableCommitFailedException.
+ assertTrue(commitResult.isDone());
+ assertFutureThrows(commitResult, RetriableCommitFailedException.class);
+ }
+
+ @Test
+ public void testAutocommitEnsureOnlyOneInflightRequest() {
TopicPartition t1p = new TopicPartition("topic1", 0);
subscriptionState.assignFromUser(singleton(t1p));
+ subscriptionState.seek(t1p, 100);
CommitRequestManager commitRequestManger = create(true, 100);
time.sleep(100);
commitRequestManger.updateAutoCommitTimer(time.milliseconds());
- // Nothing consumed therefore no commit request is sent
- assertPoll(0, commitRequestManger);
- time.sleep(10);
- subscriptionState.seekUnvalidated(t1p, new SubscriptionState.FetchPosition(100L));
List<NetworkClientDelegate.FutureCompletionHandler> futures = assertPoll(1, commitRequestManger);
- time.sleep(90);
+ time.sleep(100);
commitRequestManger.updateAutoCommitTimer(time.milliseconds());
- // We want to make sure we don't resend autocommit if the previous request has not been completed
+ // We want to make sure we don't resend autocommit if the previous request has not been
+ // completed, even if the interval expired
assertPoll(0, commitRequestManger);
assertEmptyPendingRequests(commitRequestManger);
// complete the unsent request and re-poll
- futures.get(0).onComplete(buildOffsetCommitClientResponse(new OffsetCommitResponse(0, new HashMap<>()), Errors.NONE));
+ futures.get(0).onComplete(buildOffsetCommitClientResponse(new OffsetCommitResponse(0, new HashMap<>())));
assertPoll(1, commitRequestManger);
}
@Test
- public void testOffsetFetchRequest_EnsureDuplicatedRequestSucceed() {
+ public void testOffsetFetchRequestEnsureDuplicatedRequestSucceed() {
CommitRequestManager commitRequestManger = create(true, 100);
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
Set<TopicPartition> partitions = new HashSet<>();
@@ -276,7 +389,7 @@ public class CommitRequestManagerTest {
@ParameterizedTest
@MethodSource("offsetFetchExceptionSupplier")
- public void testOffsetFetchRequest_ErroredRequests(final Errors error, final boolean isRetriable) {
+ public void testOffsetFetchRequestErroredRequests(final Errors error, final boolean isRetriable) {
CommitRequestManager commitRequestManger = create(true, 100);
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
@@ -300,19 +413,138 @@ public class CommitRequestManagerTest {
@ParameterizedTest
@MethodSource("offsetCommitExceptionSupplier")
- public void testOffsetCommitRequest_ErroredRequests(final Errors error, final boolean isRetriable) {
+ public void testOffsetCommitRequestErroredRequestsNotRetriedForAsyncCommit(final Errors error) {
CommitRequestManager commitRequestManger = create(true, 100);
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(new TopicPartition("topic", 1),
new OffsetAndMetadata(0));
- sendAndVerifyOffsetCommitRequests(commitRequestManger, offsets, error);
+ // Send commit request without expiration (async commit not expected to be retried).
+ CompletableFuture<Void> commitResult = commitRequestManger.addOffsetCommitRequest(offsets, Optional.empty(), false);
+ completeOffsetCommitRequestWithError(commitRequestManger, error);
+ NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ assertTrue(commitResult.isDone());
+ assertTrue(commitResult.isCompletedExceptionally());
+ if (error.exception() instanceof RetriableException) {
+ assertFutureThrows(commitResult, RetriableCommitFailedException.class);
+ }
- assertExceptionHandling(commitRequestManger, error);
+ // We expect that the request should not have been retried on this async commit.
+ assertExceptionHandling(commitRequestManger, error, false);
assertCoordinatorDisconnect(error);
}
+
+ @Test
+ public void testAsyncOffsetCommitThrowsRetriableCommitExceptionForUnhandledRetriable() {
+ CommitRequestManager commitRequestManger = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+
+ // Send commit request without expiration (async commit) that fails with retriable
+ // network exception that has no specific handling. Should fail with
+ // RetriableCommitException.
+ CompletableFuture<Void> commitResult = commitRequestManger.addOffsetCommitRequest(offsets, Optional.empty(), false);
+ completeOffsetCommitRequestWithError(commitRequestManger, Errors.NETWORK_EXCEPTION);
+ NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ assertTrue(commitResult.isDone());
+ assertTrue(commitResult.isCompletedExceptionally());
+ assertFutureThrows(commitResult, RetriableCommitFailedException.class);
+ }
+
+ @Test
+ public void testOffsetCommitSyncTimeoutNotReturnedOnPollAndFails() {
+ CommitRequestManager commitRequestManger = create(false, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(
+ new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+
+ // Send sync offset commit request that fails with retriable error.
+ Long expirationTimeMs = time.milliseconds() + retryBackoffMs * 2;
+ CompletableFuture<Void> commitResult = commitRequestManger.addOffsetCommitRequest(offsets, Optional.of(expirationTimeMs), false);
+ completeOffsetCommitRequestWithError(commitRequestManger, Errors.REQUEST_TIMED_OUT);
+
+ // Request retried after backoff, and fails with retriable again. Should not complete yet
+ // given that the request timeout hasn't expired.
+ time.sleep(retryBackoffMs);
+ completeOffsetCommitRequestWithError(commitRequestManger, Errors.REQUEST_TIMED_OUT);
+ assertFalse(commitResult.isDone());
+
+ // Sleep to expire the request timeout. Request should fail on the next poll.
+ time.sleep(retryBackoffMs);
+ NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ assertTrue(commitResult.isDone());
+ assertTrue(commitResult.isCompletedExceptionally());
+ }
+
+ /**
+ * Sync commit requests that fail with a retriable error continue to be retry while there is
+ * time. When time expires, they should fail with a TimeoutException.
+ */
+ @Test
+ public void testOffsetCommitSyncFailedWithRetriableThrowsTimeoutWhenRetryTimeExpires() {
+ CommitRequestManager commitRequestManger = create(false, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(
+ new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+
+ // Send offset commit request that fails with retriable error.
+ Long expirationTimeMs = time.milliseconds() + retryBackoffMs * 2;
+ CompletableFuture<Void> commitResult = commitRequestManger.addOffsetCommitRequest(offsets, Optional.of(expirationTimeMs), false);
+ completeOffsetCommitRequestWithError(commitRequestManger, Errors.COORDINATOR_NOT_AVAILABLE);
+
+ // Sleep to expire the request timeout. Request should fail on the next poll with a
+ // TimeoutException.
+ time.sleep(expirationTimeMs);
+ NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ assertTrue(commitResult.isDone());
+ assertTrue(commitResult.isCompletedExceptionally());
+ Throwable t = assertThrows(ExecutionException.class, () -> commitResult.get());
+ assertEquals(TimeoutException.class, t.getCause().getClass());
+ }
+
+ /**
+ * Async commit requests that fail with a retriable error are not retried, and they should fail
+ * with a RetriableCommitException.
+ */
+ @Test
+ public void testOffsetCommitAsyncFailedWithRetriableThrowsRetriableCommitException() {
+ CommitRequestManager commitRequestManger = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+
+ // Send async commit request that fails with retriable error (not expected to be retried).
+ Errors retriableError = Errors.COORDINATOR_NOT_AVAILABLE;
+ CompletableFuture<Void> commitResult = commitRequestManger.addOffsetCommitRequest(offsets, Optional.empty(), false);
+ completeOffsetCommitRequestWithError(commitRequestManger, retriableError);
+ NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ assertTrue(commitResult.isDone());
+ assertTrue(commitResult.isCompletedExceptionally());
+
+ // We expect that the request should not have been retried on this async commit.
+ assertExceptionHandling(commitRequestManger, retriableError, false);
+
+ // Request should have been completed with a RetriableCommitException
+ Throwable t = assertThrows(ExecutionException.class, () -> commitResult.get());
+ assertEquals(RetriableCommitFailedException.class, t.getCause().getClass());
+ assertEquals(retriableError.exception().getClass(), t.getCause().getCause().getClass());
+ }
+
+
@Test
public void testEnsureBackoffRetryOnOffsetCommitRequestTimeout() {
CommitRequestManager commitRequestManger = create(true, 100);
@@ -321,7 +553,8 @@ public class CommitRequestManagerTest {
Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(new TopicPartition("topic", 1),
new OffsetAndMetadata(0));
- commitRequestManger.addOffsetCommitRequest(offsets);
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
+ commitRequestManger.addOffsetCommitRequest(offsets, Optional.of(expirationTimeMs), false);
NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
assertEquals(1, res.unsentRequests.size());
res.unsentRequests.get(0).handler().onFailure(time.milliseconds(), new TimeoutException());
@@ -337,8 +570,14 @@ public class CommitRequestManagerTest {
}
}
- private void assertExceptionHandling(CommitRequestManager commitRequestManger, Errors errors) {
- long remainBackoffMs = retryBackoffMs;
+ private void assertExceptionHandling(CommitRequestManager commitRequestManger, Errors errors,
+ boolean requestShouldBeRetried) {
+ long remainBackoffMs;
+ if (requestShouldBeRetried) {
+ remainBackoffMs = retryBackoffMs;
+ } else {
+ remainBackoffMs = Long.MAX_VALUE;
+ }
switch (errors) {
case NOT_COORDINATOR:
case COORDINATOR_NOT_AVAILABLE:
@@ -348,7 +587,9 @@ public class CommitRequestManagerTest {
break;
case UNKNOWN_TOPIC_OR_PARTITION:
case COORDINATOR_LOAD_IN_PROGRESS:
- assertRetryBackOff(commitRequestManger, remainBackoffMs);
+ if (requestShouldBeRetried) {
+ assertRetryBackOff(commitRequestManger, remainBackoffMs);
+ }
break;
case GROUP_AUTHORIZATION_FAILED:
// failed
@@ -363,7 +604,11 @@ public class CommitRequestManagerTest {
assertPollDoesNotReturn(commitRequestManger, Long.MAX_VALUE);
break;
default:
- assertPollDoesNotReturn(commitRequestManger, Long.MAX_VALUE);
+ if (errors.exception() instanceof RetriableException && requestShouldBeRetried) {
+ assertRetryBackOff(commitRequestManger, remainBackoffMs);
+ } else {
+ assertPollDoesNotReturn(commitRequestManger, Long.MAX_VALUE);
+ }
}
}
@@ -381,6 +626,130 @@ public class CommitRequestManagerTest {
assertPoll(1, commitRequestManager);
}
+ // This should be the case where the OffsetFetch fails with invalid member epoch and the
+ // member already has a new epoch (ex. when member just joined the group or got a new
+ // epoch after a reconciliation). The request should just be retried with the new epoch
+ // and succeed.
+ @Test
+ public void testSyncOffsetFetchFailsWithStaleEpochAndRetriesWithNewEpoch() {
+ CommitRequestManager commitRequestManager = create(false, 100);
+ Set<TopicPartition> partitions = Collections.singleton(new TopicPartition("t1", 0));
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ // Send request that is expected to fail with invalid epoch.
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
+ commitRequestManager.addOffsetFetchRequest(partitions, expirationTimeMs);
+
+ // Mock member has new a valid epoch.
+ int newEpoch = 8;
+ String memberId = "member1";
+ commitRequestManager.onMemberEpochUpdated(Optional.of(newEpoch), Optional.of(memberId));
+
+ // Receive error when member already has a newer member epoch. Request should be retried.
+ completeOffsetFetchRequestWithError(commitRequestManager, partitions, Errors.STALE_MEMBER_EPOCH);
+
+ // Check that the request that failed was removed from the inflight requests buffer.
+ assertEquals(0, commitRequestManager.pendingRequests.inflightOffsetFetches.size());
+ assertEquals(1, commitRequestManager.pendingRequests.unsentOffsetFetches.size());
+
+ // Request should be retried with backoff.
+ NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ time.sleep(retryBackoffMs);
+ res = commitRequestManager.poll(time.milliseconds());
+ assertEquals(1, res.unsentRequests.size());
+
+ // The retried request should include the latest member ID and epoch.
+ OffsetFetchRequestData reqData = (OffsetFetchRequestData) res.unsentRequests.get(0).requestBuilder().build().data();
+ assertEquals(1, reqData.groups().size());
+ assertEquals(newEpoch, reqData.groups().get(0).memberEpoch());
+ assertEquals(memberId, reqData.groups().get(0).memberId());
+ }
+
+ // This should be the case of an OffsetFetch that fails because the member is not in the
+ // group anymore (left the group, failed with fatal error, or got fenced). In that case the
+ // request should fail without retry.
+ @Test
+ public void testSyncOffsetFetchFailsWithStaleEpochAndNotRetriedIfMemberNotInGroupAnymore() {
+ CommitRequestManager commitRequestManager = create(false, 100);
+ Set<TopicPartition> partitions = Collections.singleton(new TopicPartition("t1", 0));
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ // Send request that is expected to fail with invalid epoch.
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> requestResult =
+ commitRequestManager.addOffsetFetchRequest(partitions, expirationTimeMs);
+
+ // Mock member not having a valid epoch anymore (left/failed/fenced).
+ commitRequestManager.onMemberEpochUpdated(Optional.empty(), Optional.empty());
+
+ // Receive error when member is not in the group anymore. Request should fail.
+ completeOffsetFetchRequestWithError(commitRequestManager, partitions, Errors.STALE_MEMBER_EPOCH);
+
+ assertTrue(requestResult.isDone());
+ assertTrue(requestResult.isCompletedExceptionally());
+
+ // No new request should be generated on the next poll.
+ NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ }
+
+ // This should be the case where the OffsetCommit expected to be retried on
+ // STALE_MEMBER_EPOCH (ex. triggered from the reconciliation process), fails with invalid
+ // member epoch. The expectation is that if the member already has a new epoch, the request
+ // should be retried with the new epoch.
+ @Test
+ public void testOffsetCommitFailsWithStaleEpochAndRetriesWithNewEpoch() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
+
+ Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(new TopicPartition("topic", 1),
+ new OffsetAndMetadata(0));
+
+ // Send commit request expected to be retried on STALE_MEMBER_EPOCH.
+ // errors while it does not expire).
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
+ commitRequestManager.addOffsetCommitRequest(offsets, Optional.of(expirationTimeMs), true);
+
+ // Mock member has new a valid epoch.
+ int newEpoch = 8;
+ String memberId = "member1";
+ commitRequestManager.onMemberEpochUpdated(Optional.of(newEpoch), Optional.of(memberId));
+
+ completeOffsetCommitRequestWithError(commitRequestManager, Errors.STALE_MEMBER_EPOCH);
+
+ // Check that the request that failed was removed from the inflight requests buffer.
+ assertEquals(0, commitRequestManager.pendingRequests.inflightOffsetFetches.size());
+ assertEquals(1, commitRequestManager.pendingRequests.unsentOffsetCommits.size());
+
+ // Request should be retried with backoff.
+ NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
+ assertEquals(0, res.unsentRequests.size());
+ time.sleep(retryBackoffMs);
+ res = commitRequestManager.poll(time.milliseconds());
+ assertEquals(1, res.unsentRequests.size());
+
+ // The retried request should include the latest member ID and epoch.
+ OffsetCommitRequestData reqData = (OffsetCommitRequestData) res.unsentRequests.get(0).requestBuilder().build().data();
+ assertEquals(newEpoch, reqData.generationIdOrMemberEpoch());
+ assertEquals(memberId, reqData.memberId());
+ }
+
+ private void completeOffsetFetchRequestWithError(CommitRequestManager commitRequestManager,
+ Set<TopicPartition> partitions,
+ Errors error) {
+ NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
+ assertEquals(1, res.unsentRequests.size());
+ res.unsentRequests.get(0).future().complete(buildOffsetFetchClientResponse(res.unsentRequests.get(0), partitions, error));
+ }
+
+ private void completeOffsetCommitRequestWithError(CommitRequestManager commitRequestManager,
+ Errors error) {
+ NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
+ assertEquals(1, res.unsentRequests.size());
+ res.unsentRequests.get(0).future().complete(mockOffsetCommitResponse("topic", 1, (short) 1, error));
+ }
+
private void testRetriable(final CommitRequestManager commitRequestManger,
final List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures) {
futures.forEach(f -> assertFalse(f.isDone()));
@@ -395,20 +764,25 @@ public class CommitRequestManagerTest {
futures.forEach(f -> assertTrue(f.isCompletedExceptionally()));
}
- // Supplies (error, isRetriable)
+ /**
+ * @return {@link Errors} that could be received in OffsetCommit responses.
+ */
private static Stream<Arguments> offsetCommitExceptionSupplier() {
return Stream.of(
- Arguments.of(Errors.NOT_COORDINATOR, true),
- Arguments.of(Errors.COORDINATOR_LOAD_IN_PROGRESS, true),
- Arguments.of(Errors.UNKNOWN_SERVER_ERROR, false),
- Arguments.of(Errors.GROUP_AUTHORIZATION_FAILED, false),
- Arguments.of(Errors.OFFSET_METADATA_TOO_LARGE, false),
- Arguments.of(Errors.INVALID_COMMIT_OFFSET_SIZE, false),
- Arguments.of(Errors.UNKNOWN_TOPIC_OR_PARTITION, true),
- Arguments.of(Errors.COORDINATOR_NOT_AVAILABLE, true),
- Arguments.of(Errors.REQUEST_TIMED_OUT, true),
- Arguments.of(Errors.FENCED_INSTANCE_ID, false),
- Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED, false));
+ Arguments.of(Errors.NOT_COORDINATOR),
+ Arguments.of(Errors.COORDINATOR_LOAD_IN_PROGRESS),
+ Arguments.of(Errors.UNKNOWN_SERVER_ERROR),
+ Arguments.of(Errors.GROUP_AUTHORIZATION_FAILED),
+ Arguments.of(Errors.OFFSET_METADATA_TOO_LARGE),
+ Arguments.of(Errors.INVALID_COMMIT_OFFSET_SIZE),
+ Arguments.of(Errors.UNKNOWN_TOPIC_OR_PARTITION),
+ Arguments.of(Errors.COORDINATOR_NOT_AVAILABLE),
+ Arguments.of(Errors.REQUEST_TIMED_OUT),
+ Arguments.of(Errors.FENCED_INSTANCE_ID),
+ Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED),
+ Arguments.of(Errors.NETWORK_EXCEPTION),
+ Arguments.of(Errors.STALE_MEMBER_EPOCH),
+ Arguments.of(Errors.UNKNOWN_MEMBER_ID));
}
// Supplies (error, isRetriable)
@@ -425,12 +799,16 @@ public class CommitRequestManagerTest {
Arguments.of(Errors.COORDINATOR_NOT_AVAILABLE, false),
Arguments.of(Errors.REQUEST_TIMED_OUT, false),
Arguments.of(Errors.FENCED_INSTANCE_ID, false),
- Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED, false));
+ Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED, false),
+ Arguments.of(Errors.UNKNOWN_MEMBER_ID, false),
+ // Adding STALE_MEMBER_EPOCH as non-retriable here because it is only retried if a new
+ // member epoch is received. Tested separately.
+ Arguments.of(Errors.STALE_MEMBER_EPOCH, false));
}
@ParameterizedTest
@MethodSource("partitionDataErrorSupplier")
- public void testOffsetFetchRequest_PartitionDataError(final Errors error, final boolean isRetriable) {
+ public void testOffsetFetchRequestPartitionDataError(final Errors error, final boolean isRetriable) {
CommitRequestManager commitRequestManger = create(true, 100);
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
Set<TopicPartition> partitions = new HashSet<>();
@@ -438,8 +816,9 @@ public class CommitRequestManagerTest {
TopicPartition tp2 = new TopicPartition("t2", 3);
partitions.add(tp1);
partitions.add(tp2);
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future =
- commitRequestManger.addOffsetFetchRequest(partitions);
+ commitRequestManger.addOffsetFetchRequest(partitions, expirationTimeMs);
NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
assertEquals(1, res.unsentRequests.size());
@@ -467,7 +846,7 @@ public class CommitRequestManagerTest {
Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(new TopicPartition("topic", 1),
new OffsetAndMetadata(0));
- commitRequestManger.addOffsetCommitRequest(offsets);
+ commitRequestManger.addOffsetCommitRequest(offsets, Optional.empty(), false);
commitRequestManger.signalClose();
NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
assertEquals(1, res.unsentRequests.size());
@@ -496,9 +875,9 @@ public class CommitRequestManagerTest {
int numRequest,
final Errors error) {
List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = new ArrayList<>();
-
+ long expirationTimeMs = time.milliseconds() + defaultApiTimeoutMs;
for (int i = 0; i < numRequest; i++) {
- futures.add(commitRequestManger.addOffsetFetchRequest(partitions));
+ futures.add(commitRequestManger.addOffsetFetchRequest(partitions, expirationTimeMs));
}
NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
@@ -510,16 +889,21 @@ public class CommitRequestManagerTest {
return futures;
}
- private void sendAndVerifyOffsetCommitRequests(
+ private void sendAndVerifyOffsetCommitRequestFailedAndMaybeRetried(
final CommitRequestManager commitRequestManger,
- final Map<TopicPartition, OffsetAndMetadata> offsets,
- final Errors error) {
- commitRequestManger.addOffsetCommitRequest(offsets);
+ final Errors error,
+ final CompletableFuture<Void> commitResult) {
+ completeOffsetCommitRequestWithError(commitRequestManger, error);
NetworkClientDelegate.PollResult res = commitRequestManger.poll(time.milliseconds());
- assertEquals(1, res.unsentRequests.size());
- res.unsentRequests.get(0).future().complete(mockOffsetCommitResponse("topic", 1, (short) 1, error));
- res = commitRequestManger.poll(time.milliseconds());
assertEquals(0, res.unsentRequests.size());
+ if (error.exception() instanceof RetriableException) {
+ // Commit should not complete if the timer is still valid and the error is retriable.
+ assertFalse(commitResult.isDone());
+ } else {
+ // Commit should fail if the timer expired or the error is not retriable.
+ assertTrue(commitResult.isDone());
+ assertTrue(commitResult.isCompletedExceptionally());
+ }
}
private List<NetworkClientDelegate.FutureCompletionHandler> assertPoll(
@@ -556,11 +940,11 @@ public class CommitRequestManagerTest {
this.subscriptionState,
new ConsumerConfig(props),
this.coordinatorRequestManager,
- this.backgroundEventHandler,
- this.groupState,
+ DEFAULT_GROUP_ID,
+ Optional.of(DEFAULT_GROUP_INSTANCE_ID),
retryBackoffMs,
retryBackoffMaxMs,
- 0));
+ OptionalDouble.of(0)));
}
private ClientResponse buildOffsetFetchClientResponse(
@@ -576,10 +960,7 @@ public class CommitRequestManagerTest {
return buildOffsetFetchClientResponse(request, topicPartitionData, error);
}
- private ClientResponse buildOffsetCommitClientResponse(final OffsetCommitResponse commitResponse,
- final Errors error) {
- OffsetCommitResponseData data = new OffsetCommitResponseData();
- OffsetCommitResponse response = new OffsetCommitResponse(data);
+ private ClientResponse buildOffsetCommitClientResponse(final OffsetCommitResponse commitResponse) {
short apiVersion = 1;
return new ClientResponse(
new RequestHeader(ApiKeys.OFFSET_COMMIT, apiVersion, "", 1),
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java
index e6bca05d8a1..4ff652ff493 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java
@@ -135,7 +135,7 @@ public class ConsumerNetworkThreadTest {
@Test
public void testApplicationEvent() {
- ApplicationEvent e = new CommitApplicationEvent(new HashMap<>());
+ ApplicationEvent e = new CommitApplicationEvent(new HashMap<>(), Optional.empty());
applicationEventsQueue.add(e);
consumerNetworkThread.runOnce();
verify(applicationEventProcessor, times(1)).process(e);
@@ -151,7 +151,7 @@ public class ConsumerNetworkThreadTest {
@Test
public void testCommitEvent() {
- ApplicationEvent e = new CommitApplicationEvent(new HashMap<>());
+ ApplicationEvent e = new CommitApplicationEvent(new HashMap<>(), Optional.empty());
applicationEventsQueue.add(e);
consumerNetworkThread.runOnce();
verify(applicationEventProcessor).process(any(CommitApplicationEvent.class));
@@ -208,7 +208,8 @@ public class ConsumerNetworkThreadTest {
verify(applicationEventProcessor).process(any(AssignmentChangeApplicationEvent.class));
verify(networkClient, times(1)).poll(anyLong(), anyLong());
verify(commitRequestManager, times(1)).updateAutoCommitTimer(currentTimeMs);
- verify(commitRequestManager, times(1)).maybeAutoCommit(offset);
+ // Assignment change should generate an async commit (not retried).
+ verify(commitRequestManager, times(1)).maybeAutoCommitAllConsumedAsync();
}
@Test
@@ -272,8 +273,8 @@ public class ConsumerNetworkThreadTest {
coordinatorRequestManager.markCoordinatorUnknown("test", time.milliseconds());
client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "group-id", node));
prepareOffsetCommitRequest(new HashMap<>(), Errors.NONE, false);
- CompletableApplicationEvent<Void> event1 = spy(new CommitApplicationEvent(Collections.emptyMap()));
- ApplicationEvent event2 = new CommitApplicationEvent(Collections.emptyMap());
+ CompletableApplicationEvent<Void> event1 = spy(new CommitApplicationEvent(Collections.emptyMap(), Optional.empty()));
+ ApplicationEvent event2 = new CommitApplicationEvent(Collections.emptyMap(), Optional.empty());
CompletableFuture<Void> future = new CompletableFuture<>();
when(event1.future()).thenReturn(future);
applicationEventsQueue.add(event1);
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerTestBuilder.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerTestBuilder.java
index eb3cdd0926e..013ddca9d54 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerTestBuilder.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerTestBuilder.java
@@ -108,15 +108,14 @@ public class ConsumerTestBuilder implements Closeable {
this.backgroundEventQueue = new LinkedBlockingQueue<>();
this.backgroundEventHandler = spy(new BackgroundEventHandler(logContext, backgroundEventQueue));
GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
- 100,
- DEFAULT_MAX_POLL_INTERVAL_MS,
- DEFAULT_HEARTBEAT_INTERVAL_MS,
- groupInfo.map(gi -> gi.groupState.groupId).orElse(null),
- groupInfo.flatMap(gi -> gi.groupState.groupInstanceId),
- DEFAULT_RETRY_BACKOFF_MS,
- DEFAULT_RETRY_BACKOFF_MAX_MS,
- true);
- GroupState groupState = new GroupState(groupRebalanceConfig);
+ 100,
+ DEFAULT_MAX_POLL_INTERVAL_MS,
+ DEFAULT_HEARTBEAT_INTERVAL_MS,
+ groupInfo.map(gi -> gi.groupId).orElse(null),
+ groupInfo.flatMap(gi -> gi.groupInstanceId),
+ DEFAULT_RETRY_BACKOFF_MS,
+ DEFAULT_RETRY_BACKOFF_MAX_MS,
+ true);
ApiVersions apiVersions = new ApiVersions();
Properties properties = new Properties();
@@ -130,8 +129,8 @@ public class ConsumerTestBuilder implements Closeable {
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
groupInfo.ifPresent(gi -> {
- properties.put(GROUP_ID_CONFIG, gi.groupState.groupId);
- gi.groupState.groupInstanceId.ifPresent(groupInstanceId -> properties.put(GROUP_INSTANCE_ID_CONFIG, groupInstanceId));
+ properties.put(GROUP_ID_CONFIG, gi.groupId);
+ gi.groupInstanceId.ifPresent(groupInstanceId -> properties.put(GROUP_INSTANCE_ID_CONFIG, groupInstanceId));
});
this.config = new ConsumerConfig(properties);
@@ -182,19 +181,20 @@ public class ConsumerTestBuilder implements Closeable {
DEFAULT_RETRY_BACKOFF_MS,
DEFAULT_RETRY_BACKOFF_MAX_MS,
backgroundEventHandler,
- gi.groupState.groupId
+ gi.groupId
));
CommitRequestManager commit = spy(new CommitRequestManager(time,
logContext,
subscriptions,
config,
coordinator,
- backgroundEventHandler,
- groupState));
+ gi.groupId,
+ gi.groupInstanceId));
MembershipManager mm = spy(
new MembershipManagerImpl(
- gi.groupState.groupId,
- gi.groupState.groupInstanceId,
+ gi.groupId,
+ gi.groupInstanceId,
+ groupRebalanceConfig.rebalanceTimeoutMs,
gi.serverAssignor,
subscriptions,
commit,
@@ -309,25 +309,27 @@ public class ConsumerTestBuilder implements Closeable {
}
public static class GroupInformation {
-
- final GroupState groupState;
+ final String groupId;
+ final Optional<String> groupInstanceId;
final int heartbeatIntervalMs;
final double heartbeatJitterMs;
final Optional<String> serverAssignor;
- public GroupInformation(GroupState groupState) {
- this(groupState, DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_HEARTBEAT_JITTER_MS, Optional.of(DEFAULT_REMOTE_ASSIGNOR));
+ public GroupInformation(String groupId, Optional<String> groupInstanceId) {
+ this(groupId, groupInstanceId, DEFAULT_HEARTBEAT_INTERVAL_MS, DEFAULT_HEARTBEAT_JITTER_MS,
+ Optional.of(DEFAULT_REMOTE_ASSIGNOR));
}
- public GroupInformation(GroupState groupState, int heartbeatIntervalMs, double heartbeatJitterMs, Optional<String> serverAssignor) {
- this.groupState = groupState;
+ public GroupInformation(String groupId, Optional<String> groupInstanceId, int heartbeatIntervalMs, double heartbeatJitterMs, Optional<String> serverAssignor) {
this.heartbeatIntervalMs = heartbeatIntervalMs;
this.heartbeatJitterMs = heartbeatJitterMs;
this.serverAssignor = serverAssignor;
+ this.groupId = groupId;
+ this.groupInstanceId = groupInstanceId;
}
}
static Optional<GroupInformation> createDefaultGroupInformation() {
- return Optional.of(new GroupInformation(new GroupState(DEFAULT_GROUP_ID, Optional.empty())));
+ return Optional.of(new GroupInformation(DEFAULT_GROUP_ID, Optional.empty()));
}
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
index 2d64c9ee2e7..3c07dab6a7f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
@@ -121,7 +121,8 @@ public class HeartbeatRequestManagerTest {
cleanup();
ConsumerTestBuilder.GroupInformation gi = new ConsumerTestBuilder.GroupInformation(
- new GroupState(DEFAULT_GROUP_ID, groupInstanceId),
+ DEFAULT_GROUP_ID,
+ groupInstanceId,
0,
0.0,
Optional.of(DEFAULT_REMOTE_ASSIGNOR)
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
index dbd627e9d0c..d8f187ebab4 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
@@ -66,6 +66,7 @@ import static org.mockito.ArgumentMatchers.anyCollection;
import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
@@ -76,6 +77,7 @@ public class MembershipManagerImplTest {
private static final String GROUP_ID = "test-group";
private static final String MEMBER_ID = "test-member-1";
+ private static final int REBALANCE_TIMEOUT = 100;
private static final int MEMBER_EPOCH = 1;
private final LogContext logContext = new LogContext();
@@ -107,17 +109,18 @@ public class MembershipManagerImplTest {
private MembershipManagerImpl createMembershipManagerJoiningGroup() {
MembershipManagerImpl manager = spy(new MembershipManagerImpl(
- GROUP_ID, Optional.empty(), Optional.empty(), subscriptionState, commitRequestManager,
- metadata, logContext, Optional.empty(), backgroundEventHandler));
+ GROUP_ID, Optional.empty(), REBALANCE_TIMEOUT, Optional.empty(),
+ subscriptionState, commitRequestManager, metadata, logContext, Optional.empty(),
+ backgroundEventHandler));
manager.transitionToJoining();
return manager;
}
private MembershipManagerImpl createMembershipManagerJoiningGroup(String groupInstanceId) {
MembershipManagerImpl manager = spy(new MembershipManagerImpl(
- GROUP_ID, Optional.ofNullable(groupInstanceId), Optional.empty(),
- subscriptionState, commitRequestManager, metadata, logContext,
- Optional.empty(), backgroundEventHandler));
+ GROUP_ID, Optional.ofNullable(groupInstanceId), REBALANCE_TIMEOUT, Optional.empty(),
+ subscriptionState, commitRequestManager, metadata, logContext, Optional.empty(),
+ backgroundEventHandler));
manager.transitionToJoining();
return manager;
}
@@ -125,9 +128,9 @@ public class MembershipManagerImplTest {
private MembershipManagerImpl createMembershipManagerJoiningGroup(String groupInstanceId,
String serverAssignor) {
MembershipManagerImpl manager = new MembershipManagerImpl(
- GROUP_ID, Optional.ofNullable(groupInstanceId), Optional.ofNullable(serverAssignor),
- subscriptionState, commitRequestManager, metadata, logContext, Optional.empty(),
- backgroundEventHandler);
+ GROUP_ID, Optional.ofNullable(groupInstanceId), REBALANCE_TIMEOUT,
+ Optional.ofNullable(serverAssignor), subscriptionState, commitRequestManager,
+ metadata, logContext, Optional.empty(), backgroundEventHandler);
manager.transitionToJoining();
return manager;
}
@@ -151,8 +154,9 @@ public class MembershipManagerImplTest {
public void testMembershipManagerRegistersForClusterMetadataUpdatesOnFirstJoin() {
// First join should register to get metadata updates
MembershipManagerImpl manager = new MembershipManagerImpl(
- GROUP_ID, Optional.empty(), Optional.empty(), subscriptionState, commitRequestManager,
- metadata, logContext, Optional.empty(), backgroundEventHandler);
+ GROUP_ID, Optional.empty(), REBALANCE_TIMEOUT, Optional.empty(),
+ subscriptionState, commitRequestManager, metadata, logContext, Optional.empty(),
+ backgroundEventHandler);
manager.transitionToJoining();
verify(metadata).addClusterUpdateListener(manager);
clearInvocations(metadata);
@@ -229,8 +233,9 @@ public class MembershipManagerImplTest {
@Test
public void testTransitionToFailedWhenTryingToJoin() {
MembershipManagerImpl membershipManager = new MembershipManagerImpl(
- GROUP_ID, Optional.empty(), Optional.empty(), subscriptionState, commitRequestManager, metadata,
- logContext, Optional.empty(), backgroundEventHandler);
+ GROUP_ID, Optional.empty(), REBALANCE_TIMEOUT, Optional.empty(),
+ subscriptionState, commitRequestManager, metadata, logContext, Optional.empty(),
+ backgroundEventHandler);
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
membershipManager.transitionToJoining();
@@ -246,6 +251,66 @@ public class MembershipManagerImplTest {
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
}
+ @Test
+ public void testListenersGetNotifiedOnTransitionsToFatal() {
+ MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup();
+ MemberStateListener listener = mock(MemberStateListener.class);
+ membershipManager.registerStateListener(listener);
+ mockStableMember(membershipManager);
+ verify(listener).onMemberEpochUpdated(Optional.of(MEMBER_EPOCH), Optional.of(MEMBER_ID));
+ clearInvocations(listener);
+
+ // Transition to FAILED before getting member ID/epoch
+ membershipManager.transitionToFatal();
+ assertEquals(MemberState.FATAL, membershipManager.state());
+ verify(listener).onMemberEpochUpdated(Optional.empty(), Optional.empty());
+ }
+
+ @Test
+ public void testListenersGetNotifiedOnTransitionsToLeavingGroup() {
+ MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup();
+ MemberStateListener listener = mock(MemberStateListener.class);
+ membershipManager.registerStateListener(listener);
+ mockStableMember(membershipManager);
+ verify(listener).onMemberEpochUpdated(Optional.of(MEMBER_EPOCH), Optional.of(MEMBER_ID));
+ clearInvocations(listener);
+
+ mockLeaveGroup();
+ membershipManager.leaveGroup();
+ assertEquals(MemberState.LEAVING, membershipManager.state());
+ verify(listener).onMemberEpochUpdated(Optional.empty(), Optional.empty());
+ }
+
+ @Test
+ public void testListenersGetNotifiedOfMemberEpochUpdatesOnlyIfItChanges() {
+ MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup();
+ MemberStateListener listener = mock(MemberStateListener.class);
+ membershipManager.registerStateListener(listener);
+ int epoch = 5;
+
+ membershipManager.onHeartbeatResponseReceived(new ConsumerGroupHeartbeatResponseData()
+ .setErrorCode(Errors.NONE.code())
+ .setMemberId(MEMBER_ID)
+ .setMemberEpoch(epoch));
+
+ verify(listener).onMemberEpochUpdated(Optional.of(epoch), Optional.of(MEMBER_ID));
+ clearInvocations(listener);
+
+ membershipManager.onHeartbeatResponseReceived(new ConsumerGroupHeartbeatResponseData()
+ .setErrorCode(Errors.NONE.code())
+ .setMemberId(MEMBER_ID)
+ .setMemberEpoch(epoch));
+ verify(listener, never()).onMemberEpochUpdated(any(), any());
+ }
+
+ private void mockStableMember(MembershipManagerImpl membershipManager) {
+ ConsumerGroupHeartbeatResponse heartbeatResponse = createConsumerGroupHeartbeatResponse(null);
+ membershipManager.onHeartbeatResponseReceived(heartbeatResponse.data());
+ assertEquals(MemberState.STABLE, membershipManager.state());
+ assertEquals(MEMBER_ID, membershipManager.memberId());
+ assertEquals(MEMBER_EPOCH, membershipManager.memberEpoch());
+ }
+
@Test
public void testFencingWhenStateIsReconciling() {
MembershipManager membershipManager = mockJoinAndReceiveAssignment(false);
@@ -265,25 +330,22 @@ public class MembershipManagerImplTest {
@Test
public void testFencingWhenStateIsPrepareLeaving() {
MembershipManagerImpl membershipManager = createMemberInStableState();
-
- // Start leaving group, blocked waiting for commit of all consumed to complete.
- CompletableFuture<Void> commitResult = mockPrepareLeavingStuckCommitting();
- membershipManager.leaveGroup();
+ ConsumerRebalanceListenerInvoker invoker = consumerRebalanceListenerInvoker();
+ ConsumerRebalanceListenerCallbackCompletedEvent callbackEvent =
+ mockPrepareLeavingStuckOnUserCallback(membershipManager, invoker);
assertEquals(MemberState.PREPARE_LEAVING, membershipManager.state());
// Get fenced while preparing to leave the group. Member should ignore the fence
// (no callbacks or rejoin) and should transition to UNSUBSCRIBED to
// effectively stop sending heartbeats.
- MockRebalanceListener rebalanceListener = new MockRebalanceListener();
- when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(rebalanceListener));
+ clearInvocations(subscriptionState);
membershipManager.transitionToFenced();
+ testFenceIsNoOp(membershipManager);
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
- testFenceIsNoOp(membershipManager, rebalanceListener);
// When PREPARE_LEAVING completes, the member should remain UNSUBSCRIBED (no last HB sent
// because member is already out of the group in the broker).
- when(subscriptionState.rebalanceListener()).thenReturn(Optional.empty());
- commitResult.complete(null);
+ completeCallback(callbackEvent, membershipManager);
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
assertTrue(membershipManager.shouldSkipHeartbeat());
}
@@ -291,10 +353,9 @@ public class MembershipManagerImplTest {
@Test
public void testNewAssignmentIgnoredWhenStateIsPrepareLeaving() {
MembershipManagerImpl membershipManager = createMemberInStableState();
-
- // Start leaving group, blocked waiting for commit of all consumed to complete.
- CompletableFuture<Void> commitResult = mockPrepareLeavingStuckCommitting();
- membershipManager.leaveGroup();
+ ConsumerRebalanceListenerInvoker invoker = consumerRebalanceListenerInvoker();
+ ConsumerRebalanceListenerCallbackCompletedEvent callbackEvent =
+ mockPrepareLeavingStuckOnUserCallback(membershipManager, invoker);
assertEquals(MemberState.PREPARE_LEAVING, membershipManager.state());
// Get new assignment while preparing to leave the group. Member should continue leaving
@@ -308,9 +369,8 @@ public class MembershipManagerImplTest {
assertTrue(membershipManager.topicsWaitingForMetadata().isEmpty());
verify(membershipManager, never()).markReconciliationInProgress();
- // When commit completes member should transition to LEAVING.
- when(subscriptionState.rebalanceListener()).thenReturn(Optional.empty());
- commitResult.complete(null);
+ // When callback completes member should transition to LEAVING.
+ completeCallback(callbackEvent, membershipManager);
assertEquals(MemberState.LEAVING, membershipManager.state());
}
@@ -325,10 +385,9 @@ public class MembershipManagerImplTest {
// Get fenced while leaving. Member should not trigger any callback or try to
// rejoin, and should continue leaving the group as it was before getting fenced.
- MockRebalanceListener rebalanceListener = new MockRebalanceListener();
- when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(rebalanceListener));
+ clearInvocations(subscriptionState);
membershipManager.transitionToFenced();
- testFenceIsNoOp(membershipManager, rebalanceListener);
+ testFenceIsNoOp(membershipManager);
assertEquals(MemberState.LEAVING, membershipManager.state());
membershipManager.onHeartbeatRequestSent();
@@ -336,8 +395,9 @@ public class MembershipManagerImplTest {
// Last heartbeat sent is expected to fail, leading to a call to transitionToFenced. That
// should be no-op because the member already left.
+ clearInvocations(subscriptionState);
membershipManager.transitionToFenced();
- testFenceIsNoOp(membershipManager, rebalanceListener);
+ testFenceIsNoOp(membershipManager);
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
}
@@ -606,16 +666,15 @@ public class MembershipManagerImplTest {
@Test
public void testFatalFailureWhenStateIsPrepareLeaving() {
MembershipManagerImpl membershipManager = createMemberInStableState();
-
- // Start leaving group, blocked waiting for commit of all consumed to complete.
- CompletableFuture<Void> commitResult = mockPrepareLeavingStuckCommitting();
- membershipManager.leaveGroup();
+ ConsumerRebalanceListenerInvoker invoker = consumerRebalanceListenerInvoker();
+ ConsumerRebalanceListenerCallbackCompletedEvent callbackEvent =
+ mockPrepareLeavingStuckOnUserCallback(membershipManager, invoker);
assertEquals(MemberState.PREPARE_LEAVING, membershipManager.state());
testStateUpdateOnFatalFailure(membershipManager);
- // When commit completes member should abort the leave operation and remain in FATAL state.
- commitResult.complete(null);
+ // When callback completes member should abort the leave operation and remain in FATAL.
+ completeCallback(callbackEvent, membershipManager);
assertEquals(MemberState.FATAL, membershipManager.state());
}
@@ -1103,7 +1162,8 @@ public class MembershipManagerImplTest {
membershipManager,
invoker,
ConsumerRebalanceListenerMethodName.ON_PARTITIONS_ASSIGNED,
- topicPartitions(topicName, 0, 1)
+ topicPartitions(topicName, 0, 1),
+ true
);
assertFalse(membershipManager.reconciliationInProgress());
@@ -1128,7 +1188,8 @@ public class MembershipManagerImplTest {
membershipManager,
invoker,
ConsumerRebalanceListenerMethodName.ON_PARTITIONS_REVOKED,
- topicPartitions(topicName, 0, 1)
+ topicPartitions(topicName, 0, 1),
+ true
);
assertTrue(membershipManager.reconciliationInProgress());
@@ -1137,7 +1198,8 @@ public class MembershipManagerImplTest {
membershipManager,
invoker,
ConsumerRebalanceListenerMethodName.ON_PARTITIONS_ASSIGNED,
- Collections.emptySortedSet()
+ Collections.emptySortedSet(),
+ true
);
assertFalse(membershipManager.reconciliationInProgress());
@@ -1185,7 +1247,8 @@ public class MembershipManagerImplTest {
membershipManager,
invoker,
ConsumerRebalanceListenerMethodName.ON_PARTITIONS_REVOKED,
- topicPartitions(topicName, 0)
+ topicPartitions(topicName, 0),
+ true
);
assertFalse(membershipManager.reconciliationInProgress());
@@ -1233,7 +1296,8 @@ public class MembershipManagerImplTest {
membershipManager,
invoker,
ConsumerRebalanceListenerMethodName.ON_PARTITIONS_REVOKED,
- topicPartitions("topic1", 0)
+ topicPartitions("topic1", 0),
+ true
);
assertTrue(membershipManager.reconciliationInProgress());
@@ -1243,7 +1307,8 @@ public class MembershipManagerImplTest {
membershipManager,
invoker,
ConsumerRebalanceListenerMethodName.ON_PARTITIONS_ASSIGNED,
- Collections.emptySortedSet()
+ Collections.emptySortedSet(),
+ true
);
assertFalse(membershipManager.reconciliationInProgress());
@@ -1302,7 +1367,8 @@ public class MembershipManagerImplTest {
membershipManager,
invoker,
ConsumerRebalanceListenerMethodName.ON_PARTITIONS_LOST,
- topicPartitions("topic1", 0)
+ topicPartitions("topic1", 0),
+ true
);
// Step 4: Receive ack and make sure we're done and our listener was called appropriately
@@ -1358,10 +1424,11 @@ public class MembershipManagerImplTest {
return Collections.singletonMap(topicId, topicIdPartitions);
}
- private void performCallback(MembershipManagerImpl membershipManager,
+ private ConsumerRebalanceListenerCallbackCompletedEvent performCallback(MembershipManagerImpl membershipManager,
ConsumerRebalanceListenerInvoker invoker,
ConsumerRebalanceListenerMethodName expectedMethodName,
- SortedSet<TopicPartition> expectedPartitions) {
+ SortedSet<TopicPartition> expectedPartitions,
+ boolean complete) {
// We expect only our enqueued event in the background queue.
assertEquals(1, backgroundEventQueue.size());
assertNotNull(backgroundEventQueue.peek());
@@ -1377,14 +1444,21 @@ public class MembershipManagerImplTest {
neededEvent.partitions(),
neededEvent.future()
);
- membershipManager.consumerRebalanceListenerCallbackCompleted(invokedEvent);
+
+ if (complete) {
+ completeCallback(invokedEvent, membershipManager);
+ }
+ return invokedEvent;
+ }
+
+ private void completeCallback(ConsumerRebalanceListenerCallbackCompletedEvent callbackCompletedEvent,
+ MembershipManagerImpl membershipManager) {
+ membershipManager.consumerRebalanceListenerCallbackCompleted(callbackCompletedEvent);
}
- private void testFenceIsNoOp(MembershipManagerImpl membershipManager,
- MockRebalanceListener rebalanceListener) {
+ private void testFenceIsNoOp(MembershipManagerImpl membershipManager) {
assertNotEquals(0, membershipManager.memberEpoch());
- assertEquals(0, rebalanceListener.lostCount);
- assertEquals(0, rebalanceListener.revokedCount);
+ verify(subscriptionState, never()).rebalanceListener();
}
@Test
@@ -1506,7 +1580,7 @@ public class MembershipManagerImplTest {
if (withAutoCommit) {
when(commitRequestManager.autoCommitEnabled()).thenReturn(true);
CompletableFuture<Void> commitResult = new CompletableFuture<>();
- when(commitRequestManager.maybeAutoCommitAllConsumed()).thenReturn(commitResult);
+ when(commitRequestManager.maybeAutoCommitAllConsumedNow(any(), anyBoolean())).thenReturn(commitResult);
return commitResult;
} else {
return CompletableFuture.completedFuture(null);
@@ -1680,17 +1754,27 @@ public class MembershipManagerImplTest {
doNothing().when(subscriptionState).markPendingRevocation(anySet());
}
- private CompletableFuture<Void> mockPrepareLeavingStuckCommitting() {
+ private ConsumerRebalanceListenerCallbackCompletedEvent mockPrepareLeavingStuckOnUserCallback(
+ MembershipManagerImpl membershipManager,
+ ConsumerRebalanceListenerInvoker invoker) {
String topicName = "topic1";
TopicPartition ownedPartition = new TopicPartition(topicName, 0);
+
+ // Start leaving group, blocked waiting for callback to complete.
+ CounterConsumerRebalanceListener listener = new CounterConsumerRebalanceListener();
when(subscriptionState.assignedPartitions()).thenReturn(Collections.singleton(ownedPartition));
when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
- when(subscriptionState.rebalanceListener()).thenReturn(Optional.empty());
+ when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(listener));
doNothing().when(subscriptionState).markPendingRevocation(anySet());
- when(commitRequestManager.autoCommitEnabled()).thenReturn(true);
- CompletableFuture<Void> commitResult = new CompletableFuture<>();
- when(commitRequestManager.maybeAutoCommitAllConsumed()).thenReturn(commitResult);
- return commitResult;
+ when(commitRequestManager.autoCommitEnabled()).thenReturn(false);
+ membershipManager.leaveGroup();
+ return performCallback(
+ membershipManager,
+ invoker,
+ ConsumerRebalanceListenerMethodName.ON_PARTITIONS_REVOKED,
+ topicPartitions(ownedPartition.topic(), ownedPartition.partition()),
+ false
+ );
}
private void testStateUpdateOnFatalFailure(MembershipManagerImpl membershipManager) {
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventProcessorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
similarity index 77%
rename from clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventProcessorTest.java
rename to clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
index 7074542c3b0..075552fcd55 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventProcessorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
@@ -14,11 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.kafka.clients.consumer.internals;
+package org.apache.kafka.clients.consumer.internals.events;
-import org.apache.kafka.clients.consumer.internals.events.ApplicationEventProcessor;
-import org.apache.kafka.clients.consumer.internals.events.CommitOnCloseApplicationEvent;
-import org.apache.kafka.clients.consumer.internals.events.LeaveOnCloseApplicationEvent;
+import org.apache.kafka.clients.consumer.internals.CommitRequestManager;
+import org.apache.kafka.clients.consumer.internals.ConsumerMetadata;
+import org.apache.kafka.clients.consumer.internals.CoordinatorRequestManager;
+import org.apache.kafka.clients.consumer.internals.FetchRequestManager;
+import org.apache.kafka.clients.consumer.internals.HeartbeatRequestManager;
+import org.apache.kafka.clients.consumer.internals.MembershipManager;
+import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate;
+import org.apache.kafka.clients.consumer.internals.OffsetsRequestManager;
+import org.apache.kafka.clients.consumer.internals.RequestManagers;
+import org.apache.kafka.clients.consumer.internals.TopicMetadataRequestManager;
import org.apache.kafka.common.utils.LogContext;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -29,6 +36,7 @@ import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -86,6 +94,15 @@ public class ApplicationEventProcessorTest {
verify(commitRequestManager).signalClose();
}
+ @Test
+ public void testExpirationCalculation() {
+ assertEquals(Long.MAX_VALUE, processor.getExpirationTimeForTimeout(Long.MAX_VALUE));
+ assertEquals(Long.MAX_VALUE, processor.getExpirationTimeForTimeout(Long.MAX_VALUE - 1));
+ long timeout = processor.getExpirationTimeForTimeout(1000);
+ assertTrue(timeout > 0);
+ assertTrue(timeout < Long.MAX_VALUE);
+ }
+
@Test
public void testPrepClosingLeaveGroupEvent() {
LeaveOnCloseApplicationEvent event = new LeaveOnCloseApplicationEvent();