diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/k8s/TopicManager.kt b/theodolite-quarkus/src/main/kotlin/theodolite/k8s/TopicManager.kt index a956f8218499491fc91e1595692aa8ddcd3912d0..e82a133b3e5439e72987f3db107f4e81a1d01cd5 100644 --- a/theodolite-quarkus/src/main/kotlin/theodolite/k8s/TopicManager.kt +++ b/theodolite-quarkus/src/main/kotlin/theodolite/k8s/TopicManager.kt @@ -30,7 +30,7 @@ class TopicManager(private val kafkaConfig: HashMap<String, Any>) { result.all().get()// wait for the future object } catch (e: Exception) { - removeTopics(newTopics.map { topic -> topic.name() }) + delete(newTopics.map { topic -> topic.name() }, kafkaAdmin) logger.warn { "Error during topic creation." } logger.debug { e } logger.warn { "Will retry the topic creation after 2 seconds" } @@ -54,6 +54,11 @@ class TopicManager(private val kafkaConfig: HashMap<String, Any>) { */ fun removeTopics(topics: List<String>) { val kafkaAdmin: AdminClient = AdminClient.create(this.kafkaConfig) + delete(topics, kafkaAdmin) + kafkaAdmin.close() + } + + private fun delete(topics: List<String>, kafkaAdmin: AdminClient) { var deleted = false while (!deleted) { @@ -82,6 +87,6 @@ class TopicManager(private val kafkaConfig: HashMap<String, Any>) { sleep(RETRY_TIME) } } - kafkaAdmin.close() } + }