tests

Author: lianetm

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerAssignTest.java

@@ -38,6 +38,7 @@
 import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
 import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG;
 import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -298,6 +299,58 @@ private void testAssignAndRetrievingCommittedOffsetsMultipleTimes(GroupProtocol
         }
     }
 
+    @ClusterTest
+    public void testAsyncPollAfterTopicDeleted() throws Exception {
+        testPollAfterTopicDeleted(GroupProtocol.CONSUMER);
+    }
+
+    @ClusterTest
+    public void testClassicPollAfterTopicDeleted() throws Exception {
+        testPollAfterTopicDeleted(GroupProtocol.CLASSIC);
+    }
+
+    /**
+     * Test that poll() doesn't fail to retrieve committed offsets after the assigned topic is deleted.
+     * Validates fix for KAFKA-20165.
+     */
+    private void testPollAfterTopicDeleted(GroupProtocol groupProtocol) throws Exception {
+        String topicToDelete = "topic-to-delete-assign";
+        clusterInstance.createTopic(topicToDelete, 1, (short) BROKER_COUNT);
+        TopicPartition tpToDelete = new TopicPartition(topicToDelete, 0);
+
+        int numRecords = 10;
+        long startingTimestamp = System.currentTimeMillis();
+
+        Map<String, Object> consumerConfig = Map.of(
+            GROUP_PROTOCOL_CONFIG, groupProtocol.name,
+            GROUP_ID_CONFIG, "test-group",
+            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+        try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer(consumerConfig);
+             var admin = clusterInstance.admin()) {
+
+            // Send records and consume them (this caches the topic ID in the consumer)
+            ClientsTestUtils.sendRecords(clusterInstance, tpToDelete, numRecords, startingTimestamp);
+            consumer.assign(List.of(tpToDelete));
+            consumer.seek(tpToDelete, 0);
+            ClientsTestUtils.consumeAndVerifyRecords(consumer, tpToDelete, numRecords, 0, 0, startingTimestamp);
+            consumer.commitSync();
+
+            // Delete the topic and wait for deletion to propagate
+            admin.deleteTopics(List.of(topicToDelete)).all().get();
+            Thread.sleep(1000);
+
+            // Change assignment to force the consumer to fetch committed offsets on next poll()
+            // The consumer still has the topic ID cached, so it will use version 10+
+            // and get UNKNOWN_TOPIC_ID from the broker
+            consumer.unsubscribe();
+            consumer.assign(List.of(tpToDelete));
+
+            // poll() should not throw - internally fetches committed offsets for deleted topic and recovers
+            assertDoesNotThrow(() -> consumer.poll(java.time.Duration.ofMillis(5000)));
+        }
+    }
+
     private static class CountConsumerCommitCallback implements OffsetCommitCallback {
         int successCount = 0;
         int failCount = 0;

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java

@@ -208,6 +208,67 @@ private void testNoCommittedOffsets(GroupProtocol groupProtocol) {
         }
     }
 
+    /**
+     * Test that calling consumer.committed() for a deleted topic returns null for each partition.
+     * This validates the fix for KAFKA-20165.
+     */
+    @ClusterTest
+    public void testClassicConsumerCommittedDeletedTopic() throws Exception {
+        testConsumerCommittedDeletedTopic(GroupProtocol.CLASSIC);
+    }
+
+    /**
+     * Test that calling consumer.committed() for a deleted topic returns null for each partition.
+     * This validates the fix for KAFKA-20165.
+     */
+    @ClusterTest
+    public void testAsyncConsumerCommittedDeletedTopic() throws Exception {
+        testConsumerCommittedDeletedTopic(GroupProtocol.CONSUMER);
+    }
+
+    /**
+     * Common test logic for both Classic and Async consumer.
+     * Tests that calling consumer.committed() for a deleted topic returns null.
+     * As per the Javadoc for committed(): "If any of the partitions requested do not exist, the result
+     * map will contain null as the value for that partition."
+     *
+     * For Classic consumer (topic names only), the GroupCoordinator returns no error for deleted topics (offsets -1).
+     *
+     * For Async consumer (topic IDs and topic names), the GroupCoordinator returns UNKNOWN_TOPIC_ID for the deleted topic
+     * when the client uses topic IDs.
+     * The consumer handles this as a retriable partition error and eventually returns null,
+     * keeping the same contract as the ClassicConsumer.
+     */
+    private void testConsumerCommittedDeletedTopic(GroupProtocol groupProtocol) throws Exception {
+        String topicToDelete = "topic-to-delete";
+        cluster.createTopic(topicToDelete, 1, (short) BROKER_COUNT);
+        TopicPartition tpToDelete = new TopicPartition(topicToDelete, 0);
+
+        try (var consumer = createConsumer(groupProtocol, false);
+             var admin = cluster.admin()) {
+            consumer.assign(List.of(tpToDelete));
+
+            // Commit an offset to ensure the consumer has the topic ID cached and there's data to fetch
+            consumer.commitSync(Map.of(tpToDelete, new OffsetAndMetadata(0)));
+
+            // Verify the commit was successful
+            assertEquals(0, consumer.committed(Set.of(tpToDelete)).get(tpToDelete).offset());
+
+            // Delete the topic
+            admin.deleteTopics(List.of(topicToDelete)).all().get();
+
+            // Eventually, the response should return null for the deleted topic partition.
+            TestUtils.waitForCondition(
+                () -> {
+                    var committed = consumer.committed(Set.of(tpToDelete), Duration.ofMillis(5000));
+                    return committed.get(tpToDelete) == null;
+                },
+                10000,
+                "Expected null for deleted topic partition"
+            );
+        }
+    }
+
     @ClusterTest
     public void testClassicConsumerAsyncCommit() throws InterruptedException {
         testAsyncCommit(GroupProtocol.CLASSIC);