Skip to content

KAFKA-16235: Skip deleted-topic auto-commit retries before rebalance #23213

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MdTanwer wants to merge 1 commit into
base: trunk
Choose a base branch
from
+131 −3
Open

KAFKA-16235: Skip deleted-topic auto-commit retries before rebalance #23213

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 40 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
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.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.errors.UnstableOffsetCommitException;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.message.JoinGroupRequestData;
Expand Down Expand Up @@ -782,11 +783,16 @@ protected boolean onJoinPrepare(Timer timer, int generation, String memberId) {
// 1. if joinPrepareTimer has expired
// 2. if offset commit failed with non-retriable exception
// 3. if offset commit success
// 4. if offset commit failed because a topic/partition was deleted (UNKNOWN_TOPIC_OR_PARTITION).
// Retrying that error blocks JoinGroup indefinitely and, with the EAGER protocol, pauses
// fetches for every assigned partition (KAFKA-16235).
boolean onJoinPrepareAsyncCommitCompleted = true;
if (joinPrepareTimer.isExpired()) {
log.error("Asynchronous auto-commit of offsets failed: joinPrepare timeout. Will continue to join group");
} else if (!autoCommitOffsetRequestFuture.isDone()) {
onJoinPrepareAsyncCommitCompleted = false;
} else if (autoCommitOffsetRequestFuture.failed() && isUnknownTopicOrPartition(autoCommitOffsetRequestFuture.exception())) {
log.info("Asynchronous auto-commit of offsets failed because a topic or partition was deleted. Will continue to join group.");
} else if (autoCommitOffsetRequestFuture.failed() && autoCommitOffsetRequestFuture.isRetriable()) {
log.debug("Asynchronous auto-commit of offsets failed with retryable error: {}. Will retry it.",
autoCommitOffsetRequestFuture.exception().getMessage());
Expand Down Expand Up @@ -1183,7 +1189,7 @@ public boolean commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets,

private void maybeAutoCommitOffsetsSync(Timer timer) {
if (autoCommitEnabled) {
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptions.allConsumed();
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = offsetsForExistingTopics(subscriptions.allConsumed());
try {
log.debug("Sending synchronous auto-commit of offsets {}", allConsumedOffsets);
if (!commitOffsetsSync(allConsumedOffsets, timer))
Expand Down Expand Up @@ -1230,8 +1236,40 @@ private boolean invokePendingAsyncCommits(Timer timer) {
return false;
}

/**
* For group management, drop offsets for topics that are no longer in the subscription
* (for example a regex match that disappeared after a topic delete). Committing those
* partitions returns {@link Errors#UNKNOWN_TOPIC_OR_PARTITION}, which is retriable and
* can stall rebalance (KAFKA-16235). Manual assignment is left unchanged.
*/
private Map<TopicPartition, OffsetAndMetadata> offsetsForExistingTopics(Map<TopicPartition, OffsetAndMetadata> offsets) {
if (!subscriptions.hasAutoAssignedPartitions()) {
return offsets;
}
Map<TopicPartition, OffsetAndMetadata> filtered = new HashMap<>(offsets.size());
for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
if (subscriptions.subscription().contains(entry.getKey().topic())) {
filtered.put(entry.getKey(), entry.getValue());
} else {
log.debug("Skipping offset commit for partition {} because its topic is no longer subscribed",
entry.getKey());
}
}
return filtered;
}

private static boolean isUnknownTopicOrPartition(Throwable exception) {
while (exception != null) {
if (exception instanceof UnknownTopicOrPartitionException) {
return true;
}
exception = exception.getCause();
}
return false;
}

private RequestFuture<Void> autoCommitOffsetsAsync() {
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptions.allConsumed();
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = offsetsForExistingTopics(subscriptions.allConsumed());
log.debug("Sending asynchronous auto-commit of offsets {}", allConsumedOffsets);

return commitOffsetsAsync(allConsumedOffsets, (offsets, exception) -> {
Expand Down
92 changes: 91 additions & 1 deletion ...ts/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1359,14 +1359,56 @@ rebalanceConfig, new Metrics(), assignors, true, subscriptions)) {
}
}

@Test
public void testPatternSubscriptionDeletedTopicDoesNotBlockRebalanceAutoCommit() {
try (ConsumerCoordinator coordinator = buildCoordinator(
rebalanceConfig, new Metrics(), assignors, true, subscriptions)) {
subscriptions.setRebalanceListener(rebalanceListener, rebalanceConsumer);
subscriptions.subscribe(Pattern.compile("test.*"));
client.updateMetadata(RequestTestUtils.metadataUpdateWith(1, Map.of(
topic1, 1,
topic2, 1
)));
coordinator.maybeUpdateSubscriptionMetadata();
assertEquals(Set.of(topic1, topic2), subscriptions.subscription());

client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

partitionAssignor.prepare(singletonMap(consumerId, Arrays.asList(t1p, t2p)));
client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE));
client.prepareResponse(syncGroupResponse(Arrays.asList(t1p, t2p), Errors.NONE));
coordinator.poll(time.timer(Long.MAX_VALUE));
subscriptions.seek(t1p, 10);
subscriptions.seek(t2p, 20);

MetadataResponse deletedMetadataResponse = RequestTestUtils.metadataUpdateWith(1, Map.of(topic1, 1));
client.updateMetadata(deletedMetadataResponse);
coordinator.maybeUpdateSubscriptionMetadata();
client.prepareMetadataUpdate(deletedMetadataResponse);

// Deleted topic2 must not be included in the rebalance auto-commit.
prepareOffsetCommitRequest(singletonMap(t1p, 10L), Errors.NONE);
partitionAssignor.prepare(singletonMap(consumerId, singletonList(t1p)));
client.prepareResponse(joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE));
client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE));

// A short timer would expire if UNKNOWN_TOPIC_OR_PARTITION were retried until rebalance.timeout.ms.
coordinator.poll(time.timer(1000L));

assertEquals(singleton(topic1), subscriptions.subscription());
assertFalse(coordinator.rejoinNeededOrPending());
}
}

@Test
public void testOnJoinPrepareWithOffsetCommitShouldSuccessAfterRetry() {
try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, true, Optional.empty(), false)) {
int generationId = 42;
String memberId = "consumer-42";

Timer pollTimer = time.timer(100L);
client.prepareResponse(offsetCommitResponse(singletonMap(t1p, Errors.UNKNOWN_TOPIC_OR_PARTITION)));
client.prepareResponse(offsetCommitResponse(singletonMap(t1p, Errors.COORDINATOR_LOAD_IN_PROGRESS)));
boolean res = coordinator.onJoinPrepare(pollTimer, generationId, memberId);
assertFalse(res);

Expand All @@ -1381,6 +1423,54 @@ public void testOnJoinPrepareWithOffsetCommitShouldSuccessAfterRetry() {
}
}

@Test
public void testOnJoinPrepareContinuesWhenOffsetCommitUnknownTopicOrPartition() {
try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, true, Optional.empty(), false)) {
int generationId = 42;
String memberId = "consumer-42";

Timer pollTimer = time.timer(100L);
client.prepareResponse(offsetCommitResponse(singletonMap(t1p, Errors.UNKNOWN_TOPIC_OR_PARTITION)));
boolean res = coordinator.onJoinPrepare(pollTimer, generationId, memberId);
assertTrue(res, "onJoinPrepare should continue joining when a committed partition no longer exists");

assertFalse(client.hasPendingResponses());
assertFalse(client.hasInFlightRequests());
assertFalse(coordinator.coordinatorUnknown());
}
}

@Test
public void testOnJoinPrepareSkipsAutoCommitForUnsubscribedTopics() {
try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, true, Optional.empty(), false)) {
subscriptions.subscribe(Set.of());
assertTrue(subscriptions.subscription().isEmpty());

boolean res = coordinator.onJoinPrepare(time.timer(100L), 42, "consumer-42");
assertTrue(res);
assertFalse(client.hasInFlightRequests());
assertFalse(client.hasPendingResponses());
}
}

@Test
public void testOnJoinPrepareAutoCommitOmitsTopicsMissingFromMetadata() {
try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, true, Optional.empty(), false)) {
subscriptions.assignFromSubscribed(Set.of(t1p, t2p));
subscriptions.seek(t1p, 100);
subscriptions.seek(t2p, 200);
client.updateMetadata(RequestTestUtils.metadataUpdateWith(1, Map.of(topic1, 1)));

prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.NONE);

boolean res = coordinator.onJoinPrepare(time.timer(1000L), 42, "consumer-42");
assertTrue(res);
assertFalse(client.hasPendingResponses());
assertFalse(client.hasInFlightRequests());
assertFalse(coordinator.coordinatorUnknown());
}
}

@Test
public void testOnJoinPrepareWithOffsetCommitShouldKeepJoinAfterNonRetryableException() {
try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, true, Optional.empty(), false)) {
Expand Down