spring-kafka-consumer: seek to beginning example

Listener.kt

class MyListener() : ConsumerSeekAware

override fun onPartitionsAssigned(
            assignments: MutableMap<TopicPartition, Long>?,
            callback: ConsumerSeekAware.ConsumerSeekCallback?
    ) {
        logger.info { "on partitions assigned: ${assignments?.toMap()}" }

        if (fromBeginning && !soughtToBeginning.get()) {
            val assignedTopicPartitions = assignments?.map { it.key } ?: error("no assignments found")
            if (callback == null) error("seek callback not found")

            assignedTopicPartitions.forEach { topicPartition ->
                callback.seekToBeginning(topicPartition.topic(), topicPartition.partition())
            }
            logger.warn { "Kafka reset partition offset to beginning ! sought to beginning" }
            soughtToBeginning.set(true)
        }
    }

    override fun registerSeekCallback(callback: ConsumerSeekAware.ConsumerSeekCallback?):Unit =
        logger.info { "Kafka Seek" }

    override fun onIdleContainer(
            assignments: MutableMap<TopicPartition, Long>?, 
            callback: ConsumerSeekAware.ConsumerSeekCallback?
    ):Unit = logger.info { "Kafka Idle" }