You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ca...@apache.org on 2024/03/25 19:59:48 UTC
(kafka) branch trunk updated: KAFKA-16224: Do not retry committing if topic or partition deleted (#15581)
This is an automated email from the ASF dual-hosted git repository.
cadonna pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new da6bb365bc6 KAFKA-16224: Do not retry committing if topic or partition deleted (#15581)
da6bb365bc6 is described below
commit da6bb365bc68a5d46ab37963429c39d972e011cd
Author: Bruno Cadonna <ca...@apache.org>
AuthorDate: Mon Mar 25 20:59:41 2024 +0100
KAFKA-16224: Do not retry committing if topic or partition deleted (#15581)
Current logic for auto-committing offsets when partitions are revoked
will retry continuously when getting UNKNOWN_TOPIC_OR_PARTITION,
leading to the member not completing the revocation in time.
This commit considers error UNKNOWN_TOPIC_OR_PARTITION to be fatal
in the context of an auto-commit of offsets before a revocation,
even though the error is defined as retriable. This ensures that
the revocation can finish in time.
Reviewers: Andrew Schofield <as...@confluent.io>, Lucas Brutschy <lb...@confluent.io>, Lianet Magrans <li...@gmail.com>
---
.../consumer/internals/CommitRequestManager.java | 24 ++++++++++++++--------
.../consumer/internals/MembershipManagerImpl.java | 2 +-
.../internals/CommitRequestManagerTest.java | 4 ++--
.../internals/MembershipManagerImplTest.java | 4 ++--
4 files changed, 21 insertions(+), 13 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
index f7acbde60bd..67ea6eb3134 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
@@ -30,6 +30,7 @@ import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.StaleMemberEpochException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.errors.UnstableOffsetCommitException;
import org.apache.kafka.common.message.OffsetCommitRequestData;
import org.apache.kafka.common.message.OffsetCommitResponseData;
@@ -272,14 +273,18 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
* Commit consumed offsets if auto-commit is enabled, regardless of the auto-commit interval.
* This is used for committing offsets before revoking partitions. This will retry committing
* the latest offsets until the request succeeds, fails with a fatal error, or the timeout
- * expires. Note that this considers {@link Errors#STALE_MEMBER_EPOCH} as a retriable error,
+ * expires. Note that:
+ * - Considers {@link Errors#STALE_MEMBER_EPOCH} as a retriable error,
* and will retry it including the latest member ID and epoch received from the broker.
+ * - Considers {@link Errors#UNKNOWN_TOPIC_OR_PARTITION} as a fatal error, and will not retry
+ * it although the error extends RetriableException. The reason is that if a topic or partition
+ * is deleted, revocation would not finish in time since the auto commit would keep retrying.
*
* @return Future that will complete when the offsets are successfully committed. It will
* complete exceptionally if the commit fails with a non-retriable error, or if the retry
* timeout expires.
*/
- public CompletableFuture<Void> maybeAutoCommitSyncNow(final long retryExpirationTimeMs) {
+ public CompletableFuture<Void> maybeAutoCommitSyncBeforeRevocation(final long retryExpirationTimeMs) {
if (!autoCommitEnabled()) {
return CompletableFuture.completedFuture(null);
}
@@ -287,12 +292,12 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
CompletableFuture<Void> result = new CompletableFuture<>();
OffsetCommitRequestState requestState =
createOffsetCommitRequest(subscriptions.allConsumed(), Optional.of(retryExpirationTimeMs));
- autoCommitSyncNowWithRetries(requestState, result);
+ autoCommitSyncBeforeRevocationWithRetries(requestState, result);
return result;
}
- private void autoCommitSyncNowWithRetries(OffsetCommitRequestState requestAttempt,
- CompletableFuture<Void> result) {
+ private void autoCommitSyncBeforeRevocationWithRetries(OffsetCommitRequestState requestAttempt,
+ CompletableFuture<Void> result) {
CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> commitAttempt = requestAutoCommit(requestAttempt);
commitAttempt.whenComplete((committedOffsets, error) -> {
if (error == null) {
@@ -300,16 +305,19 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
} else {
if (error instanceof RetriableException || isStaleEpochErrorAndValidEpochAvailable(error)) {
if (error instanceof TimeoutException && requestAttempt.isExpired) {
- log.debug("Auto-commit sync timed out and won't be retried anymore");
+ log.debug("Auto-commit sync before revocation timed out and won't be retried anymore");
+ result.completeExceptionally(error);
+ } else if (error instanceof UnknownTopicOrPartitionException) {
+ log.debug("Auto-commit sync before revocation failed because topic or partition were deleted");
result.completeExceptionally(error);
} else {
// Make sure the auto-commit is retries with the latest offsets
requestAttempt.offsets = subscriptions.allConsumed();
requestAttempt.resetFuture();
- autoCommitSyncNowWithRetries(requestAttempt, result);
+ autoCommitSyncBeforeRevocationWithRetries(requestAttempt, result);
}
} else {
- log.debug("Auto-commit sync failed with non-retriable error", error);
+ log.debug("Auto-commit sync before revocation failed with non-retriable error", error);
result.completeExceptionally(error);
}
}
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 574db9ebcd0..47107469959 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
@@ -961,7 +961,7 @@ public class MembershipManagerImpl implements MembershipManager {
// 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.maybeAutoCommitSyncNow(getExpirationTimeForTimeout(rebalanceTimeoutMs));
+ commitResult = commitRequestManager.maybeAutoCommitSyncBeforeRevocation(getExpirationTimeForTimeout(rebalanceTimeoutMs));
// Execute commit -> onPartitionsRevoked -> onPartitionsAssigned.
commitResult.whenComplete((__, commitReqError) -> {
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 d22d6189042..e5b276adf74 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
@@ -912,7 +912,7 @@ public class CommitRequestManagerTest {
long expirationTimeMs = time.milliseconds() + retryBackoffMs * 2;
// Send commit request expected to be retried on STALE_MEMBER_EPOCH error while it does not expire
- commitRequestManager.maybeAutoCommitSyncNow(expirationTimeMs);
+ commitRequestManager.maybeAutoCommitSyncBeforeRevocation(expirationTimeMs);
int newEpoch = 8;
String memberId = "member1";
@@ -923,7 +923,7 @@ public class CommitRequestManagerTest {
completeOffsetCommitRequestWithError(commitRequestManager, error);
- if (error.exception() instanceof RetriableException || error == Errors.STALE_MEMBER_EPOCH) {
+ if ((error.exception() instanceof RetriableException || error == Errors.STALE_MEMBER_EPOCH) && error != Errors.UNKNOWN_TOPIC_OR_PARTITION) {
assertEquals(1, commitRequestManager.pendingRequests.unsentOffsetCommits.size(),
"Request to be retried should be added to the outbound queue");
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 b2c75f495dd..cef608a902d 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
@@ -597,7 +597,7 @@ public class MembershipManagerImplTest {
membershipManager.onHeartbeatSuccess(createConsumerGroupHeartbeatResponse(assignment2).data());
assertEquals(MemberState.RECONCILING, membershipManager.state());
CompletableFuture<Void> commitResult = new CompletableFuture<>();
- when(commitRequestManager.maybeAutoCommitSyncNow(anyLong())).thenReturn(commitResult);
+ when(commitRequestManager.maybeAutoCommitSyncBeforeRevocation(anyLong())).thenReturn(commitResult);
membershipManager.poll(time.milliseconds());
// Get fenced, commit completes
@@ -2457,7 +2457,7 @@ public class MembershipManagerImplTest {
if (withAutoCommit) {
when(commitRequestManager.autoCommitEnabled()).thenReturn(true);
CompletableFuture<Void> commitResult = new CompletableFuture<>();
- when(commitRequestManager.maybeAutoCommitSyncNow(anyLong())).thenReturn(commitResult);
+ when(commitRequestManager.maybeAutoCommitSyncBeforeRevocation(anyLong())).thenReturn(commitResult);
return commitResult;
} else {
return CompletableFuture.completedFuture(null);