如何使用spring kafka从给定的主题和分区以特定的偏移量重新发送（读取）旧kafka消息？

Question

给定主题名、分区号和偏移量，我如何从该主题中读取一条记录？

在基于Sprng引导的应用程序中，我使用Kafka导入业务数据。导入记录被发送到导入队列，并被一个或多个业务模块使用。即使消费者未能从记录中导入数据以继续从以下记录中导入数据，记录也会始终得到确认。

稍后，用户（在他/她修复了一些相关的业务数据后）可以决定重新发送一个或多个失败（但已确认）的导入记录。

每个记录的偏移量、分区号和主题名都存储在我的应用程序内部的SQL数据库中。

从参考文档和一些StackOverflow问题中，我发现我必须：

1. 设置容器（消费者/监听器）
2. 倒带（查找）到所需偏移量
3. 读一条记录
4. 跳过读取剩余记录

这是阅读Kafka主题中的一条旧记录的唯一方法吗？或者有更简单的解决方案吗？

正如@Gary所建议的：

ConsumerRecord<byte[], byte[]> read(String topic, int partition, long offset) {
    Map<String, Object> configs = Map.of(
            "bootstrap.servers", "localhost:9092",
            "group.id", "incubator_retry",
            "max.poll.records", 1);
    DefaultKafkaConsumerFactory<byte[], byte[]> consumerFactory = new DefaultKafkaConsumerFactory<>(
            configs, new ByteArrayDeserializer(), new ByteArrayDeserializer());

    try (Consumer<byte[], byte[]> consumer = consumerFactory.createConsumer()) {
        TopicPartition topicPartition = new TopicPartition(topic, partition);
        consumer.assign(List.of(topicPartition));
        consumer.seek(topicPartition, offset);
        ConsumerRecords<byte[], byte[]> consumerRecords = consumer.poll(Duration.ofMillis(5000));
        if (consumerRecords.isEmpty()) {
            throw new RuntimeException(String.format("Timeout polling from topic %s partition %d at offset %d",
                    topicPartition.topic(), topicPartition.partition(), offset));
        }
        return consumerRecords.iterator().next();
    }
}

Answer 1

有一个更简单的解决方案。

• 使用Default消费者工厂创建Kafka消费者（或简单地创建一个）
• 使用不同的group.id
• 将max.poll.records属性设置为1
• consumer.assign（...）所需的主题/分区
• 查找（...）到所需的偏移量
• 投票（...）直到您获得记录
• 关闭（）消费者

如果要使用任何消息转换（除了Kafka反序列化程序），则必须手动调用转换器。