通过在每次应用程序重新启动时创建新的使用者组来重置主题偏移量

Question

我有一个Kafka主题和一个消费者，在Spring云应用程序中分配了一个消费者组（必须）。作为一项要求，在每次应用程序重启时，我都需要从一开始就开始读取所有接收到的消息。这应该是通过resetOffsets属性实现的，但从这个问题可以清楚地看出，它目前不起作用。

我在kafka消费者api中发现了这个变通方法，它建议在每次重启时为消费者组分配一个新的随机名称，作为从最早开始阅读的一种方式。在Spring Cloud Stream中是否可能/推荐？如何为消费者组定义动态名称？

Answer 1

在SpringKafka中，如果您正在配置ConsumerVía配置文件，例如应用程序。yaml（而不是以编程方式）可以通过SpEL（spring表达语言）实现这一点，从而在每个执行偏移量上提供基于UUID的使用者组

# consume-all-configuration
auto-offset-reset: earliest
group: consumer-local-#{ T(java.util.UUID).randomUUID().toString() }

Answer 2

如果每次都要求应用程序从头开始重新启动，则有几个选项：

>

在使用kafka-consumer-groups.sh工具重新启动应用程序之前，您可以将提交的偏移重置为最早的（kafka.admin.消费者组ommand.scala）

重新启动后，应用程序可以查找到起始位置并手动提交偏移量0。如果您有auto。抵消重置设置为最早即使0不是有效的偏移量，它也会从头开始重新启动。

您可以使用不同的消费者组。每次的id值。在消费者配置bean中，在属性对象中插入如下内容：

properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());

最后，您是否使用委员会补偿？如果没有，只需禁用enable。汽车提交则应用程序将始终遵循自动。抵消重置设置。

选项1和2通常是首选选项，因为它们保持一致的组。id允许轻松将使用者实例添加到组并监视组。

Answer 3

是的，它也适用于SCSt，但是，正如您所说，设置随机组id有点棘手，尽管您可以在启动SpringApplication之前将其设置为System.property。

如果您直接使用spring kafka，那么很容易，只需实现ConsumerSekAware，在分配分区时，您就可以seektobegining。

但是，使用SCSt，您无法直接访问侦听器。

一种解决方法是，在启动SpringApplication之前，通过创建具有相同组id的消费者手动执行搜索。不过，如果你有多个应用实例，这会有点棘手，因为你每次可能会得到不同的分区。

我们将再次考虑解决这个问题（我只是对此发表了评论）。