Single embedded kafka as mvn/gradle plugin

[mykolapolonskyi]

Maybe will be good idea to put "embedded kafka" on to separated maven/gradle plugin

it can allow creation single kafka node per integration test phase

I`m trying to put 1.1.0 kafka to the
https://github.com/charithe/kafka-maven-plugin

It`s no a very successful or useful for now but want to ask advises pros/cons to this staff
Maybe there is way to create single kafka per tests but I miss it
Or etc

code

@Mojo(name = "start-kafka-broker")
class StartKafkaBrokerMojo(): AbstractMojo() {

    @Parameter(defaultValue = "9092")
    val kafkaPort: Int = 29092

    @Parameter(defaultValue = "2181")
    val zookeeperPort: Int = 22181

    @Throws(MojoExecutionException::class, MojoFailureException::class)
    override fun execute() {
        try {
            log.info("Starting Zookeeper on port $zookeeperPort and Kafka broker on port $kafkaPort")
            KafkaHandler.start(zookeeperPort, kafkaPort)
        } catch (e: Exception) {
            log.error("Failed to start Kafka broker", e)
            throw MojoExecutionException("Failed to start Kafka broker")
        }

    }
}

@Mojo(name = "stop-kafka-broker")
class StopKafkaBrokerMojo : AbstractMojo() {

    @Parameter(defaultValue = "29092")
    val kafkaPort: Int = 29092

    @Parameter(defaultValue = "22181")
    val zookeeperPort: Int = 22181

    @Throws(MojoExecutionException::class, MojoFailureException::class)
    override fun execute() {
        log.info("Attempting to stop the Kafka broker")
        KafkaHandler.stop()
    }
}


object KafkaHandler {
    private lateinit var zookeeper: TestingServer
    private lateinit var kafkaServer: KafkaServerStartable
    private lateinit var kafkaLogDir: Path

    private fun buildKafkaConfig(zookeeperQuorum: String, kafkaPort: Int): KafkaConfig {
        kafkaLogDir = Files.createTempDirectory("kafka_maven")

        val props = Properties().apply {
            put("port", "$kafkaPort")
            put("broker.id", "1")
            put("log.dirs", kafkaLogDir.toAbsolutePath().toString())
            put("zookeeper.connect", zookeeperQuorum)
        }
        return KafkaConfig(props)
    }


    fun start(zookeeperPort: Int, kafkaPort: Int) {
        zookeeper = TestingServer(zookeeperPort, true)
        val zookeeperConnectionString = zookeeper.getConnectString()
        val kafkaConfig = buildKafkaConfig(zookeeperConnectionString, kafkaPort)
        kafkaServer = KafkaServerStartable(kafkaConfig)
        kafkaServer.startup()
    }


    fun stop() {
        try {
            kafkaServer.shutdown()
            zookeeper.close()

            if (Files.exists(kafkaLogDir)) {
                Files.walkFileTree(kafkaLogDir, object : SimpleFileVisitor<Path>() {
                    @Throws(IOException::class)
                    override fun visitFile(file: Path, attrs: BasicFileAttributes): FileVisitResult {
                        Files.deleteIfExists(file)
                        return FileVisitResult.CONTINUE
                    }

                    @Throws(IOException::class)
                    override fun postVisitDirectory(dir: Path, exc: IOException?): FileVisitResult {
                        Files.deleteIfExists(dir)
                        return FileVisitResult.CONTINUE
                    }
                })
            }
        } catch (e: Exception) {
            e.printStackTrace()
        }
    }
}

[garyrussell]

See this issue and its related PR as well as the documentation.

[artembilan]

I would say this goes out of scope of this project.

Perhaps better to ask for such a functionality in Apache Kafka community.
Although it isn't clear why would one need to have an embedded Kafka in the plugin since it is going to be semantically the same as have a standalone Kafka Broker...

[mykolapolonskyi]

@artembilan - probably this ticket can be closed - I wold prefer use testcontainers next time for this purpose.

[artembilan]

Closed as Won't Fix.