How can I make sure that a Kafka topic has been deleted?

user_6204612 0
Viewed 57

Kafka's Admin.deleteTopics API unfortunately returns only after the request has been received - what only means that the topics are scheduled for deletion by the cluster, but not necessarily deleted now.

To show that as an example, the following code can often throw:

final var newTopic = new NewTopic("aaa", Optional.empty(), Optional.empty());
this.admin.createTopics(Collections.singleton(newTopic), opt).all().get();
this.admin.deleteTopics(Arrays.asList("aaa")).all().get();
this.admin.listTopics( ).names().get().contains("aaa"); // Returns 'false'.
this.admin.createTopics(Collections.singleton(newTopic), opt).all().get(); // <- throws

with an exception:

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: Topic 'aaa' is marked for deletion.
    at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
...
Caused by: org.apache.kafka.common.errors.TopicExistsException: Topic 'aaa' is marked for deletion.

Unfortunately Admin.listTopics() that does not help here, the topic stops being visible after the delete request is submitted.

So the question is - is there any programatic way (preferably API) that would allow us to monitor that the topic is really gone?

Kafka (both client & server) version used is 3.2.

apache-kafka java
2 Answers

So far I have not found anything interesting in API, but I have some possible ideas, unfortunately all of them are ugly.

  1. Each topic appears to have multiple metrics-beans related to their size, such as kafka.log:type=Log,name=LogStartOffset,topic=aaa,partition=0. So we could wait until they disappear. However to be fully sure we might need to check all brokers (or at least: all the replica-hosting ones).

  2. A broker instance appears to have a bean named kafka.controller:type=KafkaController,name=TopicsToDeleteCount what might be what we want, however without fine-grained control. And we still might need to check all brokers in a cluster and wait until all of them reach 0.

  3. Take a look into internal Zookeeper / KRaft structures. The interesting parts might be present at paths such as /brokers/topics and /admin/delete_topics, however this means bringing another (Zookeeper) dependency. KRaft is a similar case, we would be digging in brokers' internals.

user_62046120

Create a foo topic after delete your topics, this maybe speed up Kafka Deletion. And then create the useful topics. My test found that there is a certain effect.

public void reCreateTopics(Set<String> topics) {
    adminClient.deleteTopics(topics).all().get();
    createFooTopic4TriggeringDelete();

    // may be sleep 1s before creating
    TimeUnit.SECONDS.sleep(1);

    adminClient.createTopics(topics).all().get();
}


private void createFooTopic4TriggeringDelete()  {
    NewTopic foo = new NewTopic("foo_" + System.currentTimeMillis(), 1, (short) 1);
    adminClient.createTopics(Collections.singleton(foo)).all().get();
    adminClient.deleteTopics(Collections.singleton(foo.name())).all().get();
}
user_32228560
Related