动态路由/复制到 Kafka 2 中的多个主题

Question

Kafka 2 增加了对通过 TopicNameExtractor 接口进行动态路由的支持，该接口仅支持一个主题值。

也许我要描述的是一个糟糕的设计，但在这个阶段，我只是好奇 Kafka 流中有什么可能性。

假设每条消息都带有标签列表，有没有办法根据该标签列表将消息复制到多个主题？

Answer 1

马蒂亚斯如何提到你必须复制消息。它可以很容易地使用KStream::flatMapValues(ValueMapperWithKey ...)

示例代码如下。消息将根据tags: List<String>复制。

型：

public class Person {
    public String name;
    public List<String> tags;
    public transient String mainTag;
    public Person(String name, List<String> tags) {
        this.name = name;
        this.tags = tags;
    }
    public Person(String name, List<String> tags, String mainTag) {
        this.name = name;
        this.tags = tags;
        this.mainTag = mainTag;
    }
}

应用：

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "app1");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, PersonSerdes.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Person> input = builder.stream("input");
input.flatMapValues(((readOnlyKey, person) ->
        person.tags
                .stream()
                .map(tag -> new Person(person.name, person.tags, tag))
                .collect(Collectors.toList()))
).to((key, person, recordContext) -> person.mainTag);

Answer 2

目前这是不可能的。但是，已经有一个功能请求：https://issues.apache.org/jira/browse/KAFKA-7578

目前，如果记录

被复制并发送到多个接收器，则只能将记录写入多个输出主题。