从中间主题将KStream与KTable连接会导致异常

Question

我试图加入KStream与KTable。如果没有连接，我可以从中间主题“book属性-by-id”中阅读。

KTable的示例消息：

{key: {id: 1}
 value: {id: 1, attribute_name: "weight"}}

KStream的示例消息：

{key: {id: 1},
 value: {id: 1, book_id: 1, attribute_id: 1, value: 200}}

“最终聚合”主题的所需输出：

{key: {id: 1}, 
 value: {book_id: 1, attribute_name: "weight", value: 200}}
{key: {id: 1}, 
 value: {book_id: 1, attribute_name: "number_of_pages", value: 450}}

这是密码

    KStream<DefaultId, BookAttribute> bookAttributeStream = builder.stream(bookAttributeTopic, Consumed.with(defaultIdSerde, bookAttributeSerde));
    KStream<DefaultId, BookValueInt> bookValueIntStream = builder.stream(bookValueIntTopic, Consumed.with(defaultIdSerde, bookValueIntSerde));

    bookAttributeStream
        .selectKey((k, v) -> k.getId())
        .to("book-attribute-by-id", Produced.with(Serdes.Integer(), bookAttributeSerde));

    KTable<Integer, BookAttribute> bookAttributeByIdTable = builder.table("book-attribute-by-id", Consumed.with(Serdes.Integer(), bookAttributeSerde));

    // when the snippet below is commented out, consuming "book-attribute-by-id" works. 
    bookValueIntStream
        .selectKey((k, v) -> v.getAttribute_id())
        .join(bookAttributeByIdTable, (intValue, attribute) -> {
                System.out.println("intValue: " + intValue);
                System.out.println("attribute: " + attribute);
                return new BookAttributeValue(intValue, attribute);
            });

加入KStream时出现异常

线程“xxx-StreamThread-1”组织中出现异常。阿帕奇。Kafka。溪流。错误。TopologyBuilderException:无效的拓扑构建：未找到流线程[xxx-StreamThread-1]主题：组织上id的book属性。阿帕奇。Kafka。溪流。加工机内部。StreamPartitionAssignor$CopPartitionedTopicsValidator。验证（StreamPartitionAssignor.java:792）

Answer 1

我假设您使用的是kafka streams 1.0.0

问题是您必须为流创建输入主题。

在您的案例中，主题是：bookattributebyid，以及变量值的主题：bookAttributeTopic，bookValueIntTopic。

对于连接Kafka Streams必须确保，连接主题中的分区数量相等。当它试图获取主题的元数据时，会引发异常：book属性-by-id。

在运行应用程序之前，您必须手动创建book属性by id主题

在较新版本的kafka streams中，在验证分区数之前检查主题是否存在。