lazzexus/Kafkatorio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

only try to create topics if there are topics to be created

Browse source
This commit is contained in:
Adam 2022-01-25 00:11:38 +01:00
parent dbeb0e73dd
commit 4fc55b706f
1 changed files with 20 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

41
modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/admin/CreateTopics.kt
View file

@ -30,6 +30,8 @@ object CreateTopics {
KafkatorioPacket.PacketType.values().forEach { packetType -> KafkatorioPacket.PacketType.values().forEach { packetType ->
val currentTopics = currentTopics().map { it.name() }
val kafkatorioTopics = buildSet { val kafkatorioTopics = buildSet {
add(KafkatorioTopology.sourceTopic) add(KafkatorioTopology.sourceTopic)
when (packetType) { when (packetType) {
@ -42,33 +44,30 @@ object CreateTopics {
KafkatorioPacket.PacketType.PROTOTYPES -> KafkatorioPacket.PacketType.PROTOTYPES ->
add("kafkatorio.${packetType.name}.all") add("kafkatorio.${packetType.name}.all")
} }
} }.minus(currentTopics.toSet())
val currentTopics = currentTopics().map { it.name() } if (currentTopics.isNotEmpty()) {
val result = kafkatorioTopics logger.info("Creating ${kafkatorioTopics.size} topics")
.filterNot { it in currentTopics } val result = kafkaAdmin.createTopics { kafkatorioTopics }
.run {
logger.info("Creating $size topics") // wait for all brokers to become aware of the created topics
kafkaAdmin.createTopics { this } Thread.sleep(TimeUnit.SECONDS.toMillis(5))
result.values().forEach {
logger.info("Created topic: ${it.key} ${it.value.get()}")
} }
// wait for all brokers to become aware of the created topics Thread.sleep(TimeUnit.SECONDS.toMillis(5))
Thread.sleep(TimeUnit.SECONDS.toMillis(5))
result.values().forEach { logger.info("Listing topics")
logger.info("Created topic: ${it.key} ${it.value.get()}") kafkaAdmin.listTopics()
.listings()
.get()
.forEach {
logger.info("Topic: $it")
}
} }
Thread.sleep(TimeUnit.SECONDS.toMillis(5))
logger.info("Listing topics")
kafkaAdmin.listTopics()
.listings()
.get()
.forEach {
logger.info("Topic: $it")
}
} }
} }