Skip to content

KAFKA-20677: Enhance test coverage of testCannotSendToInternalTopic #22575

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
aravind-kvs wants to merge 1 commit into
base: trunk
Choose a base branch
from
+95 −6
Open

KAFKA-20677: Enhance test coverage of testCannotSendToInternalTopic #22575

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
101 changes: 95 additions & 6 deletions ...on-tests/src/test/java/org/apache/kafka/clients/producer/ProducerFailureHandlingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import static org.apache.kafka.clients.producer.ProducerConfig.RETRIES_CONFIG;
import static org.apache.kafka.common.config.TopicConfig.MIN_IN_SYNC_REPLICAS_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.apache.kafka.server.config.ReplicationConfigs.REPLICA_FETCH_MAX_BYTES_CONFIG;
import static org.apache.kafka.server.config.ReplicationConfigs.REPLICA_FETCH_RESPONSE_MAX_BYTES_DOC;
import static org.apache.kafka.server.config.ServerConfigs.MESSAGE_MAX_BYTES_CONFIG;
Expand Down Expand Up @@ -215,22 +216,110 @@ public void testSendAfterClosed(ClusterInstance clusterInstance) throws Interrup
assertThrows(IllegalStateException.class, () -> producer3.send(record));
}

/**
* Test that sending to an internal topic throws InvalidTopicException
* when auto.create.topics.enable=true and the internal topic already exists.
*/
@ClusterTest(serverProperties = {
@ClusterConfigProperty(key = AUTO_CREATE_TOPICS_ENABLE_CONFIG, value = "true")
})
public void testCannotSendToInternalTopicWhenAutoCreateTrueAndTopicExists(ClusterInstance clusterInstance) throws Exception {
// explicitly create __consumer_offsets to satisfy the "topic exists" precondition

@mingyen066 mingyen066 Jul 1, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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);
}

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);
}

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(InvalidTopicException.class, thrown.getCause(),
() -> "Expected InvalidTopicException but got " + thrown.getCause());
}
}

/**
* Test that sending to an internal topic throws InvalidTopicException
* when auto.create.topics.enable=false and the internal topic already exists.
*/
@ClusterTest
public void testCannotSendToInternalTopic(ClusterInstance clusterInstance) throws InterruptedException {
public void testCannotSendToInternalTopicWhenAutoCreateFalseAndTopicExists(ClusterInstance clusterInstance) throws Exception {
// explicitly create __consumer_offsets to satisfy the "topic exists" precondition
try (Admin admin = clusterInstance.admin()) {
Map<String, String> topicConfig = clusterInstance.brokers().get(0)
.groupCoordinator()
.groupMetadataTopicConfigs();
.groupCoordinator()
.groupMetadataTopicConfigs();
admin.createTopics(List.of(new NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1).configs(topicConfig)));
clusterInstance.waitTopicDeletion(Topic.GROUP_METADATA_TOPIC_NAME);
clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME, 1);
}

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(),
() -> producer.send(new ProducerRecord<>(
Topic.GROUP_METADATA_TOPIC_NAME,
"test".getBytes(),
"test".getBytes())).get());
assertInstanceOf(InvalidTopicException.class, thrown.getCause(),
() -> "Unexpected exception while sending to an invalid topic " + thrown.getCause());
() -> "Expected InvalidTopicException but got " + thrown.getCause());
}
}

/**
* Test that sending to an internal topic throws InvalidTopicException
* when auto.create.topics.enable=true and the internal topic does not exist.
* The broker should auto-create the internal topic even though the send is rejected.
*/
@ClusterTest(serverProperties = {
@ClusterConfigProperty(key = AUTO_CREATE_TOPICS_ENABLE_CONFIG, value = "true"),
@ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
})
public void testCannotSendToInternalTopicWhenAutoCreateTrueAndTopicNotExists(ClusterInstance clusterInstance) throws Exception {
clusterInstance.deleteTopic(Topic.GROUP_METADATA_TOPIC_NAME);

@mingyen066 mingyen066 Jul 1, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.


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(InvalidTopicException.class, thrown.getCause(),
() -> "Expected InvalidTopicException but got " + thrown.getCause());
}

// verify the broker auto-created the internal topic
clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME, 1);
}

/**
* Test that sending to an internal topic throws TimeoutException
* when auto.create.topics.enable=false and the internal topic does not exist.
* The broker returns UNKNOWN_TOPIC_OR_PARTITION as a recoverable error causing the producer to time out.
*/
@ClusterTest(serverProperties = {
@ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
})
public void testCannotSendToInternalTopicWhenAutoCreateFalseAndTopicNotExists(ClusterInstance clusterInstance) throws Exception {
clusterInstance.deleteTopic(Topic.GROUP_METADATA_TOPIC_NAME);

@mingyen066 mingyen066 Jul 1, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto


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(TimeoutException.class, thrown.getCause(),
() -> "Expected TimeoutException but got " + thrown.getCause());
}

// verify the internal topic was not auto-created
try (Admin admin = clusterInstance.admin()) {
assertFalse(admin.listTopics().names().get().contains(Topic.GROUP_METADATA_TOPIC_NAME));
}
}

Expand Down
Loading