diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java index fb0ffe11771d4..574c51aacd9a7 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java @@ -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; @@ -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()); @@ -1183,7 +1189,7 @@ public boolean commitOffsetsSync(Map offsets, private void maybeAutoCommitOffsetsSync(Timer timer) { if (autoCommitEnabled) { - Map allConsumedOffsets = subscriptions.allConsumed(); + Map allConsumedOffsets = offsetsForExistingTopics(subscriptions.allConsumed()); try { log.debug("Sending synchronous auto-commit of offsets {}", allConsumedOffsets); if (!commitOffsetsSync(allConsumedOffsets, timer)) @@ -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 offsetsForExistingTopics(Map offsets) { + if (!subscriptions.hasAutoAssignedPartitions()) { + return offsets; + } + Map filtered = new HashMap<>(offsets.size()); + for (Map.Entry 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 autoCommitOffsetsAsync() { - Map allConsumedOffsets = subscriptions.allConsumed(); + Map allConsumedOffsets = offsetsForExistingTopics(subscriptions.allConsumed()); log.debug("Sending asynchronous auto-commit of offsets {}", allConsumedOffsets); return commitOffsetsAsync(allConsumedOffsets, (offsets, exception) -> { diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java index 6ac647665b0bb..2b3f811e4134c 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java @@ -1359,6 +1359,48 @@ 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)) { @@ -1366,7 +1408,7 @@ public void testOnJoinPrepareWithOffsetCommitShouldSuccessAfterRetry() { 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); @@ -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)) {