Kafka Streams任务分配

4

我有一个使用单线程处理只有一个分区的主题的Kafka-Streams应用程序。

我需要同时运行多个实例来处理不同的主题。在我的当前场景中,所有主题仅有一个分区。

当我运行一个新的实例处理不同的主题时,流客户端不会在这个新应用程序中创建新任务,并且第一个实例继续处理任务0_0中的第一个主题,而第二个实例则保持等待状态,没有分配分区并且什么也不做。

我知道我正在使用只有一个分区的主题,但是如果我有两个实例和两个主题,每个主题都只有一个分区要处理,那么就有两个分区,为什么不能同时处理它们呢?

我怀疑这与 StreamsPartitionAssignor 有关,但在Kafka Streams应用程序中无法更改该分配策略:

Kafka Streams不允许使用自定义分区分配器。如果您自己设置了一个分区分配器,则它将被StreamsPartitionAssignor [1]覆盖。这是必要的,以确保在重新平衡期间尽可能将分区重新分配给同一消费者(也称为粘性)。

编辑:

应用程序的拓扑结构:

[2019-11-20 09:36:35,406] [INFO] stream-thread [avro-to-json-d07ad9ad-f4b6-4787-96cf-19c48e72ad46-StreamThread-1] Starting (org.apache.kafka.streams.processor.internals.StreamThread)
[2019-11-20 09:36:35,407] [INFO] stream-thread [avro-to-json-d07ad9ad-f4b6-4787-96cf-19c48e72ad46-StreamThread-1] State transition from CREATED to RUNNING (org.apache.kafka.streams.processor.internals.StreamThread)
[2019-11-20 09:36:35,407] [INFO] stream-client [avro-to-json-d07ad9ad-f4b6-4787-96cf-19c48e72ad46] Started Streams client (org.apache.kafka.streams.KafkaStreams)
Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [])
      --> KSTREAM-MAP-0000000001
    Processor: KSTREAM-MAP-0000000001 (stores: [])
      --> KSTREAM-MAP-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Processor: KSTREAM-MAP-0000000002 (stores: [])
      --> KSTREAM-TRANSFORM-0000000003
      <-- KSTREAM-MAP-0000000001
    Processor: KSTREAM-TRANSFORM-0000000003 (stores: [])
      --> KSTREAM-SINK-0000000004
      <-- KSTREAM-MAP-0000000002
    Sink: KSTREAM-SINK-0000000004 (extractor class: kafka.AvroToJson$$Lambda$97/741730375@957e06)
      <-- KSTREAM-TRANSFORM-0000000003
1个回答

5
如果我理解正确,您使用相同的 application.id 运行两个Streams客户端,除了输入主题之外,其他拓扑是相同的。这意味着由于不同的输入主题,您实际上运行了两个不同的Streams应用程序与您的Streams客户端。在同一个Kafka集群中,使用相同的 application.id 运行两个不同的Streams应用程序是未定义的行为,因为 application.id 需要是唯一的(请参见https://kafka.apache.org/23/documentation/#streamsconfigs)。
您可以选择以下操作之一:
1.增加其中一个主题的分区,并将该主题用作两个流应用程序的输入主题(使它们成为相同的应用程序),或者 2.更改两个应用程序中的一个的 application.id
请注意,选项1可为您提供自动负载均衡,而选项2则不能。

1
你说得完全正确。我之前没有理解流客户端和_application.id_之间的关系。我在两个客户端中使用了不同的application.id进行测试,结果如预期一样。非常感谢你的帮助! - kuro

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接