Delete topic using EmbeddedKafkaBroker

user_4478526 0
Viewed 737

I wrote a test class in which I'would test topic deletion

I found for EmbeddedKafkaBroker this method

embeddedKafkaBroker.addTopics("new-topic");

that make me able to create a new topic, but I can't find anything to delete a topic.

So, how can I test topic deletion?

spring-kafka-test embedded-kafka spring-kafka
2 Answers

See this one:

/**
 * Create an {@link AdminClient}; invoke the callback and reliably close the admin.
 * @param callback the callback.
 */
public void doWithAdmin(java.util.function.Consumer<AdminClient> callback) {

where you can use:

/**
 * This is a convenience method for {@link #deleteTopics(TopicCollection, DeleteTopicsOptions)}
 * with default options. See the overload for more details.
 * <p>
 * This operation is supported by brokers with version 0.10.1.0 or higher.
 *
 * @param topics The topic names to delete.
 * @return The DeleteTopicsResult.
 */
default DeleteTopicsResult deleteTopics(Collection<String> topics) {
user_27565470

To build on Artem's answer, this can be a bit tricky because of the asynchronicity of the events involved.

If the topic doesn't exist in the Kafka broker the describeTopics method in KafkaAdminClient throws an UnknownTopicOrPartitionException wrapped in an ExecutionException, so we can leverage that to check the topic's existence and wait for its creation or deletion.

EDIT: While the above strategy does work, the proper way to do this is to just wait for the future to complete, for example with admin.deleteTopics(List.of(topicName)).all().get()

But since the Embedded broker is not aware that the topic has been deleted, the embeddedKafkaBroker.getTopics() result doesn't get updated, so in order to find out if the topic has actually been deleted we need to use the admin.describeTopics(List.of(topicName)).all().get() method instead.

user_36924470
Related