KAFKA-19084: Port KAFKA-16224, KAFKA-16764 for ShareConsumers (#19369)

Currently for ShareConsumers, if we receive an
`UNKNOWN_TOPIC_OR_PARTITION` error code in the
`ShareAcknowledgeResponse`, then we retry sending the acknowledgements
until the timer expires.
We ideally do not want this when a topic/partition is deleted, hence
like the
`CommitRequestManager`(#15581), we
will treat this error as fatal and not retry the acknowledgements.

PR also suppresses `InvalidTopicException` during `unsubscribe()` which
was also added in the
`AsyncKafkaConsumer`(#16043). It was
later removed in the regular consumer
as they notified the background operations of metadata errors instead of
propagating them via `ErrorEvent`. `ShareConsumerImpl` however does
not require that change and it still propagates the metadata error back
to the application. So we would need to suppress this exception during
unsubscribe().

Reviewers: Andrew Schofield <[email protected]>, Sushant Mahajan <[email protected]>

Author: ShivsundarR

Diff for: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java

@@ -971,9 +971,11 @@ private void handlePartitionError(ShareAcknowledgeResponseData.PartitionData par
                                       AtomicBoolean shouldRetry) {
         if (partitionError.exception() != null) {
             boolean retry = false;
-            if (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH) {
+            if (partitionError == Errors.NOT_LEADER_OR_FOLLOWER || partitionError == Errors.FENCED_LEADER_EPOCH || partitionError == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
                 // If the leader has changed, there's no point in retrying the operation because the acquisition locks
                 // will have been released.
+                // If the topic or partition has been deleted, we do not retry the failed acknowledgements.
+                // Instead, these records will be re-delivered once they get timed out on the broker.
                 updateLeaderInfoMap(partitionData, partitionsWithUpdatedLeaderInfo, partitionError, tip.topicPartition());
             } else if (partitionError.exception() instanceof RetriableException) {
                 retry = true;

Diff for: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java

@@ -57,6 +57,7 @@
 import org.apache.kafka.common.errors.GroupAuthorizationException;
 import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.errors.InvalidGroupIdException;
+import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.common.internals.ClusterResourceListeners;
@@ -937,7 +938,7 @@ private void sendAcknowledgementsAndLeaveGroup(final Timer timer, final AtomicRe
             // If users have fatal error, they will get some exceptions in the background queue.
             // When running unsubscribe, these exceptions should be ignored, or users can't unsubscribe successfully.
             processBackgroundEvents(unsubscribeEvent.future(), timer, e -> (e instanceof GroupAuthorizationException
-                || e instanceof TopicAuthorizationException));
+                || e instanceof TopicAuthorizationException || e instanceof InvalidTopicException));
             log.info("Completed releasing assignment and leaving group to close consumer.");
         } catch (TimeoutException e) {
             log.warn("Consumer triggered an unsubscribe event to leave the group but couldn't " +

Diff for: clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManagerTest.java

@@ -656,6 +656,31 @@ public void testRetryAcknowledgements() throws InterruptedException {
         assertEquals(0, shareConsumeRequestManager.requestStates(0).getSyncRequestQueue().peek().getIncompleteAcknowledgementsCount(tip0));
     }
 
+    @ParameterizedTest
+    @EnumSource(value = Errors.class, names = {"FENCED_LEADER_EPOCH", "NOT_LEADER_OR_FOLLOWER", "UNKNOWN_TOPIC_OR_PARTITION"})
+    public void testFatalErrorsAcknowledgementResponse(Errors error) {
+        buildRequestManager();
+        shareConsumeRequestManager.setAcknowledgementCommitCallbackRegistered(true);
+
+        assignFromSubscribed(Collections.singleton(tp0));
+        sendFetchAndVerifyResponse(records, acquiredRecords, Errors.NONE);
+
+        Acknowledgements acknowledgements = getAcknowledgements(1, AcknowledgeType.ACCEPT, AcknowledgeType.ACCEPT, AcknowledgeType.REJECT);
+
+        shareConsumeRequestManager.commitAsync(Map.of(tip0, new NodeAcknowledgements(0, acknowledgements)),
+                calculateDeadlineMs(time.timer(defaultApiTimeoutMs)));
+
+        assertEquals(1, shareConsumeRequestManager.sendAcknowledgements());
+        client.prepareResponse(fullAcknowledgeResponse(tip0, error));
+        networkClientDelegate.poll(time.timer(0));
+
+        // Assert these errors are not retried even if they are retriable. They are treated as fatal and a metadata update is triggered.
+        assertEquals(0, shareConsumeRequestManager.requestStates(0).getAsyncRequest().getInFlightAcknowledgementsCount(tip0));
+        assertEquals(0, shareConsumeRequestManager.requestStates(0).getAsyncRequest().getIncompleteAcknowledgementsCount(tip0));
+        assertEquals(1, completedAcknowledgements.size());
+        assertEquals(3, completedAcknowledgements.get(0).get(tip0).size());
+    }
+
     @Test
     public void testRetryAcknowledgementsMultipleCommitAsync() {
         buildRequestManager();

Diff for: clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImplTest.java

@@ -34,6 +34,7 @@
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.InvalidGroupIdException;
+import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.common.errors.WakeupException;
@@ -331,6 +332,16 @@ public void testUnsubscribeWithTopicAuthorizationException() {
         assertDoesNotThrow(() -> consumer.close());
     }
 
+    @Test
+    public void testCloseWithInvalidTopicException() {
+        SubscriptionState subscriptions = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+        consumer = newConsumer(subscriptions);
+
+        backgroundEventQueue.add(new ErrorEvent(new InvalidTopicException(Set.of("!test-topic"))));
+        completeShareUnsubscribeApplicationEventSuccessfully(subscriptions);
+        assertDoesNotThrow(() -> consumer.close());
+    }
+
     @Test
     public void testCloseWithTopicAuthorizationException() {
         SubscriptionState subscriptions = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);