有没有办法在死信发布恢复中找到消费者群体？

Question

我有一个项目，使用了DLQ方法，在< code>@KafkaListener的任何异常中，错误将被发送到一个具有< code>error-结构的主题

这样，我们可以重试特定使用者组的消息。消息处理程序从主主题和重试主题中读取。

由于项目中有多个侦听器都使用相同的容器，因此我们将主题（主侦听器、错误侦听器和重试）放在应用程序属性中，并配置了 DeadLetter发布恢复程序以查找相应的 DLQ（注意，这是 Kotlin）：

val recoverer = DeadLetterPublishingRecoverer(template) { record, _ ->
    val dlq = kafkaProperties.topics.values.firstOrNull { it.main == record.topic() || it.retry == record.topic() }!!.dlq

    return@DeadLetterPublishingRecoverer TopicPartition(dlq, -1)
}

这一切都很好。但是，现在我想添加另一个@KafkaListener，该@KafkaListener与另一个听众听同一主题。因此，我需要为这个侦听器提供一个单独的使用者组。

但是，如果我使用与上面相同的逻辑来查找相应的DLQ主题，那么两个侦听器中的一个将使用另一个的DLQ，因为它在这里没有查看消费者组。

所以我的问题是：在Spring Kafka中，是否有任何方法可以找到错误发生在哪个消费者组（即＜code＞groupId＜/code＞）中？我看过文档和源代码，但我自己找不到。感谢任何帮助。

Answer 1

使用者的< code>group.id可通过调用< code > kafkautils . get groupid()获得(当容器线程启动时，它存储在< code>ThreadLocal中)。