KAFKA-20677: Enhance test coverage of testCannotSendToInternalTopic

[aravind-kvs]

JIRA: https://issues.apache.org/jira/browse/KAFKA-20677

Problem

The existing testCannotSendToInternalTopic test was broken. It attempted
to cover case 2 (auto.create.topics.enable=false, topic exists) but
incorrectly created and then deleted the internal topic before sending,
accidentally testing the wrong scenario.

Changes

Replaced the broken test with four separate test cases covering all scenarios
when sending to an internal topic:

1. auto.create.topics.enable=true, topic exists → InvalidTopicException
2. auto.create.topics.enable=false, topic exists → InvalidTopicException
3. auto.create.topics.enable=true, topic not exists → InvalidTopicException and topic gets auto-created
4. auto.create.topics.enable=false, topic not exists → TimeoutException and topic is NOT created

Note on Case 4

The JIRA describes case 4 as throwing UnknownTopicOrPartitionException.
However, the actual broker behavior is to return UNKNOWN_TOPIC_OR_PARTITION
as a recoverable error, causing the producer to retry until TimeoutException.
The test reflects actual broker behavior.

Test

Ran the new test cases locally and all 4 passed:

./gradlew :clients:clients-integration-tests:test --tests "org.apache.kafka.clients.producer.ProducerFailureHandlingTest.testCannotSendToInternalTopic*"

Results: 4 tests completed, 0 failed
Reviewers: Chia-Ping Tsai chia7712@gmail.com, Ming-Yen Chung mingyen066@gmail.com

[chia7712]

@aravind-kvs thanks for this patch. However, this ticket is already assigned to someone else, so it would be good to ask before opening a PR. Otherwise, it is a bit of a waste to have two talented people working on the same thing

[aravind-kvs]

@aravind-kvs thanks for this patch. However, this ticket is already assigned to someone else, so it would be good to ask before opening a PR. Otherwise, it is a bit of a waste to have two talented people working on the same thing

Thanks for the feedback!
I'll make sure to comment on the issue first before opening a PR next time. Happy to close this if the assignee is already working on it.

[mingyen066]

@aravind-kvs This was assigned to me, but since you've already opened a PR, you can take it — I've reassigned it to you. Please check with the assignee before opening a PR next time. I'll review the PR later.

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[mingyen066]

clusterInstance.deleteTopic(Topic.GROUP_METADATA_TOPIC_NAME) is a no-op here — __consumer_offsets doesn't exist yet at this point, and deleteTopic discards the result without awaiting it. Looks like leftover from the old test; safe to drop.

[mingyen066]

ditto

[mingyen066]

The four new tests repeat a lot of boilerplate — ...TrueAndTopicExists/...FalseAndTopicExists share nearly identical topic-creation code, and all four repeat the same send+assert block with only the expected exception differing. Worth extracting two helpers:

private void createInternalTopic(ClusterInstance clusterInstance) throws Exception {
    try (Admin admin = clusterInstance.admin()) {
        Map<String, String> topicConfig = clusterInstance.brokers().get(0)
                .groupCoordinator()
                .groupMetadataTopicConfigs();
        admin.createTopics(List.of(new NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1).configs(topicConfig)));
        clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME, 1);
    }
}

private void assertSendToInternalTopicFails(ClusterInstance clusterInstance, Class<? extends Throwable> expectedCause) {
    try (Producer<byte[], byte[]> producer = clusterInstance.producer(producerConfig(1))) {
        Exception thrown = assertThrows(ExecutionException.class,
                () -> producer.send(new ProducerRecord<>(
                        Topic.GROUP_METADATA_TOPIC_NAME, "test".getBytes(), "test".getBytes())).get());
        assertInstanceOf(expectedCause, thrown.getCause(),
                () -> "Expected " + expectedCause.getSimpleName() + " but got " + thrown.getCause());
    }
}

Then each test shrinks to:

@ClusterTest(serverProperties = {@ClusterConfigProperty(key = AUTO_CREATE_TOPICS_ENABLE_CONFIG, value = "true")})
public void testCannotSendToInternalTopicWhenAutoCreateTrueAndTopicExists(ClusterInstance clusterInstance) throws Exception {
    createInternalTopic(clusterInstance);
    assertSendToInternalTopicFails(clusterInstance, InvalidTopicException.class);
}