KafkanConsumer:`seekToEnd（）`不会使消费者从最新偏移量消费

Question

我有以下代码

class Consumer(val consumer: KafkaConsumer<String, ConsumerRecord<String>>) {

    fun run() {
        consumer.seekToEnd(emptyList())
        val pollDuration = 30 // seconds

        while (true) {
            val records = consumer.poll(Duration.ofSeconds(pollDuration))
            // perform record analysis and commitSync()
            }
        }
    }
}

消费者订阅的主题会不断收到记录。有时，消费者会因处理步骤而崩溃。然后，当使用者重新启动时，我希望它从主题的最新偏移量开始使用（即，忽略在使用者关闭时发布到主题的记录）。我认为seekToEnd（）方法可以确保这一点。然而，这种方法似乎毫无效果。消费者从其崩溃的偏移量开始消费。

什么是正确的方式使用asikToend（）？

编辑：使用以下配置创建消费者

fun <T> buildConsumer(valueDeserializer: String): KafkaConsumer<String, T> {
    val props = setupConfig(valueDeserializer)
    Common.setupConsumerSecurityProtocol(props)
    return createConsumer(props)
}

fun setupConfig(valueDeserializer: String): Properties {
    // Configuration setup
    val props = Properties()

    props[ConsumerConfig.GROUP_ID_CONFIG] = config.applicationId
    props[ConsumerConfig.CLIENT_ID_CONFIG] = config.kafka.clientId
    props[ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG] = config.kafka.bootstrapServers
    props[AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG] = config.kafka.schemaRegistryUrl

    props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = config.kafka.stringDeserializer
    props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = valueDeserializer
    props[KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG] = "true"

    props[ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG] = config.kafka.maxPollIntervalMs
    props[ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG] = config.kafka.sessionTimeoutMs

    props[ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG] = "false"
    props[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = "false"
    props[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "latest"

    return props
}

fun <T> createConsumer(props: Properties): KafkaConsumer<String, T> {
    val consumer = KafkaConsumer<String, T>(props)
    consumer.subscribe(listOf(config.kafka.inputTopic))
    return consumer
}

Answer 1

我找到解决办法了！

我需要添加一个虚拟轮询作为消费者初始化过程的一部分。由于有几种Kafka方法是惰性评估的，因此有必要使用虚拟轮询将分区分配给使用者。如果没有虚拟轮询，使用者将尝试查找空分区的末尾。因此，seekToEnd（）无效。

重要的是，虚拟轮询持续时间要足够长，以便分配分区。例如，消费者。轮询（（持续时间秒（1）），在程序转到下一个方法调用（即seekToEnd（））之前，分区没有得到分配的时间。

工作代码可能是这样的

class Consumer(val consumer: KafkaConsumer<String, ConsumerRecord<String>>) {

    fun run() {
        // Initialization 
        val pollDuration = 30 // seconds
        consumer.poll((Duration.ofSeconds(pollDuration)) // Dummy poll to get assigned partitions

        // Seek to end and commit new offset
        consumer.seekToEnd(emptyList())
        consumer.commitSync() 

        while (true) {
            val records = consumer.poll(Duration.ofSeconds(pollDuration))
            // perform record analysis and commitSync()
            }
        }
    }
}

Answer 2

seekToEnd方法需要关于实际分区的信息（用Kafka术语TopicPartition），您计划在该分区上让消费者从头到尾阅读。

我不熟悉静态编程语言API，但检查JavaDocs上的Kafka消费者的方法asikToend，你会看到，它要求一个集合的主题分区。

由于您当前使用的是emptyList（），所以它不会产生任何影响，就像您观察到的那样。