Kafka流处理器api允许多集群拓扑？

Question

使用kafka processor API（不是DSL）读取源主题并写入目标主题，对于单个kafka集群设置（也就是说，如果源主题和目标主题都驻留在同一集群上）来说工作很好，但是当源主题和目标主题驻留在不同的kafka集群上时，我将获得目标处理器上下文的NullPointerException

Topology
topology.addSource("mySource", "SourceTopic");
topology.addProcessor("SourceStreamProcessor",()->new SourceStreamProcessor(), "mySource");         

topology.addProcessor("TargetProcessor",()->new TargetProcessor(), "Target");
topology.addSink("sink1","OUTPUT_TOPIC1","TargetProcessor");
topology.addSink("sink2","OUTPUT_TOPIC2","TargetProcessor");


Properties sourceProcessorProps = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "SourceStreamProcessor"); // Kafka Cluster 1
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dev_Cluser_xx.org:9092");

Properties targetProcessorProps = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "targetStreamProcessor"); // Kafka Cluster 2
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "test_Cluser_xx.org:9092");

我们如何使用kafka streams处理器API从一个集群中的一个主题写到另一个集群中的另一个主题？

Answer 1

Kafka流不支持从一个Kafka集群读取并写入另一个Kafka集群。

您可以在一个集群内处理消息，然后使用Mirror Maker将其复制到另一个集群。