diff --git a/theodolite-quarkus/src/main/kotlin/theodolite/k8s/TopicManager.kt b/theodolite-quarkus/src/main/kotlin/theodolite/k8s/TopicManager.kt index 321fa9a945fade012a81ea4dbacb2c403e783411..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,9 @@ class TopicManager(private val kafkaConfig: HashMap<String, Any>) { result.all().get()// wait for the future object } catch (e: Exception) { + 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" } sleep(RETRY_TIME) retryCreation = true @@ -52,8 +54,13 @@ 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) { try { val result = kafkaAdmin.deleteTopics(topics) @@ -80,6 +87,6 @@ class TopicManager(private val kafkaConfig: HashMap<String, Any>) { sleep(RETRY_TIME) } } - kafkaAdmin.close() } + }