apache-kafka - 卡夫卡流 : use the same `application.id` to consume from multiple topics

Question

我有一个需要收听多个不同主题的应用程序；每个主题都有关于如何处理消息的单独逻辑。我曾想过为每个 KafkaStreams 实例使用相同的 kafka 属性，但我收到如下错误。

错误

java.lang.IllegalArgumentException: Assigned partition my-topic-1 for non-subscribed topic regex pattern; subscription pattern is my-other-topic

代码 ( Kotlin )

class KafkaSetup() {
    companion object {
        private val LOG = LoggerFactory.getLogger(this::class.java)
    }

    fun getProperties(): Properties {
        val properties = Properties()
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app")
        return properties
    }

    private fun listenOnMyTopic() {
        val kStreamBuilder = KStreamBuilder()
        val kStream: KStream<String, String> = kStreamBuilder.stream("my-topic")

        kStream.foreach { key, value -> LOG.info("do stuff") }

        val kafkaStreams = KafkaStreams(kStreamBuilder, getProperties())
        kafkaStreams.start()
    }

    private fun listenOnMyOtherTopic() {
        val kStreamBuilder = KStreamBuilder()
        val kStream: KStream<String, String> = kStreamBuilder.stream("my-other-topic")

        kStream.foreach { key, value -> LOG.info("do other stuff") }

        val kafkaStreams = KafkaStreams(kStreamBuilder, getProperties())
        kafkaStreams.start()
    }
}

我找到了这个 reference提示您不能使用 application.id对于多个主题，但是我发现很难找到支持它的引用文档。 documentation为 application.id状态:

An identifier for the stream processing application. Must be unique within the Kafka cluster. It is used as 1) the default client-id prefix, 2) the group-id for membership management, 3) the changelog topic prefix.

问题

这个错误是什么意思，是什么原因造成的。

鉴于您可以使用相同的 id 运行多个应用程序实例以从多个主题分区使用，那么“在 Kafka 集群中必须是唯一的”是什么意思？

您可以使用相同的 Kafka 流吗 application.id开始两个KafkaStreams列出不同的主题？如果是这样，如何？

详情:卡夫卡 0.11.0.2

Answer 1

Kafka Streams 通过分区而不是主题进行扩展。因此，如果您使用相同的 application.id 启动多个应用程序它们在订阅的输入主题和处理逻辑方面必须相同。该应用程序使用 application.id 形成一个消费者组。如 group.id因此，输入主题的不同分区被分配给不同的实例。

如果您有与 不同的主题同 逻辑，你可以订阅全部 一次主题(在您开始的每个实例中)。不过，缩放仍然基于分区。 (它基本上是您输入主题的“合并”。)

如果你想通过主题扩展和/或有不同的处理逻辑，你必须使用不同的 application.id适用于不同的 Kafka Streams 应用程序。