MessageHandler2

RenameTopicHandler.kt

class MirroringMessageHandler(
    private val arg: String = ""
) : MirrorMaker.MirrorMakerMessageHandler {

    override fun handle(record: BaseConsumerRecord?):
            MutableList<ProducerRecord<ByteArray, ByteArray>> {
        return if (record != null && mustMirroring(record)) {
            mutableListOf(
                ProducerRecord<ByteArray, ByteArray>(
                    record.topic(),
                    null,
                    record.timestamp(),
                    record.key(),
                    record.value(),
                    record.headers()
                )
            )
        } else {
            mutableListOf()
        }
    }

    private fun mustMirroring(record: BaseConsumerRecord): Boolean {
        val header = record.headers().find { it.key() == "dc-origin" }
        if (header == null) {
            record.headers().add("dc-origin", arg.toByteArray(Charsets.UTF_8))
            return true
        }
        return false
    }
}