Kafka streams-连接两个ktables会调用连接函数两次

Question

我试图加入2 KTables。

KTable<String, RecordBean> recordsTable = builder.table(Serdes.String(),
    new JsonPOJOSerde<>(RecordBean.class),
    bidTopic, RECORDS_STORE);

KTable<String, ImpressionBean> impressionsTable = builder.table(Serdes.String(),
    new JsonPOJOSerde<>(ImpressionBean.class),
    impressionTopic, IMPRESSIONS_STORE);

KTable<String, RecordBean> mergedByTxId = recordsTable
    .join(impressionsTable, merge());

合并功能非常简单，我只是将值从一个bean复制到另一个bean。

public static <K extends BidInfo, V extends BidInfo> ValueJoiner<K, V, K> merge() {
return (v1, v2) -> {
  v1.setRtbWinningBidAmount(v2.getRtbWinningBidAmount());
  return v1;
};

但由于某些原因，join函数在单个生成的记录上调用了两次。请参阅下面的流媒体/制作人配置

Properties streamsConfiguration = new Properties();
streamsConfiguration
    .put(StreamsConfig.APPLICATION_ID_CONFIG, "join-impressions");
streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());

streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, CLUSTER.zookeeperConnect());
streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, folder.newFolder("kafka-streams-tmp")
    .getAbsolutePath());

return streamsConfiguration;

生产者配置-

Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
producerConfig.put(ProducerConfig.RETRIES_CONFIG, 0);
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

return producerConfig;

接下来，我将提交每个流的单个记录。两个记录具有相同的密钥。我希望收到一条记录作为输出。

 IntegrationTestUtils.produceKeyValuesSynchronously(bidsTopic,
    Arrays.asList(new KeyValue("1", getRecordBean("1"))),
    getProducerProperties());

IntegrationTestUtils.produceKeyValuesSynchronously(impressionTopic,
    Arrays.asList(new KeyValue("1", getImpressionBean("1"))),
    getProducerProperties());

List<KeyValue<String, String>> parsedRecord =
    IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(getConsumerProperties(),
        outputTopic, 1);

但是ValueJoiner触发了2次，我得到了2条相同的输出记录，而不是一条。在触发时间内-两个流中的值都存在-我无法获取触发第二次执行的内容。

没有加入-我不能复制这种行为。我找不到任何2 ktable连接的工作示例-所以无法理解我的方法有什么问题。

添加演示相同行为的简单代码

KStreamBuilder builder = new KStreamBuilder();

KTable<String, String> first = builder.table("stream1", "storage1");
KTable<String, String> second = builder.table("stream2", "storage2");

KTable<String, String> joined = first.join(second, (value1, value2) -> value1);

joined.to("output");

KafkaStreams streams = new KafkaStreams(builder, getStreamingProperties());

streams.start();

IntegrationTestUtils.produceKeyValuesSynchronously("stream1",
    Arrays.asList(new KeyValue("1", "first stream")),
    getProducerProperties());

IntegrationTestUtils.produceKeyValuesSynchronously("stream2",
    Arrays.asList(new KeyValue("1", "second stream")),
    getProducerProperties());

List<KeyValue<String, String>> parsedRecord =
    IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(getConsumerProperties(),
        "output", 1);

Answer 1

我在两个KTable之间使用leftJoin发现了同样的行为，并在谷歌搜索后偶然发现了这篇文章。我不知道您使用的是哪个版本的Kafka流，但是在调试了汇流代码后，Kafka流2.0.1版本似乎故意在某些类型的连接中发送旧的和新的值，所以您会收到两次对ValueJoiner的调用。

看看org的实现。阿帕奇。Kafka。溪流。kstream。内部。KTableImpl#buildJoin构建连接拓扑，以及组织。阿帕奇。Kafka。溪流。kstream。内部。KTableKTableRightJoin。KTableKTableRightJoinProcessor#进程，它在运行时调度它。在某些情况下，显然要做两次。

以下是这种行为的一些背景https://issues.apache.org/jira/browse/KAFKA-2984

Answer 2

我得到了以下解释后张贴类似的问题，以汇流邮件组。

我认为这可能与缓存有关。这两个表的缓存是独立刷新的，因此有可能两次获得相同的记录。如果stream1和stream2都收到同一密钥的记录，并且缓存刷新，则：

stream1中的缓存将刷新、执行连接并生成记录。

stream2中的缓存将刷新、执行连接并生成记录。

从技术上讲，这是可以的，因为连接的结果是另一个KTable，因此KTable中的值将是正确的值。

将以下变量设置为0后，StreamsConfig。缓存\最大\字节\缓冲\配置，0-问题已解决。我仍然得到了2条记录——但现在有一条记录是用null连接的——根据上面提供的连接语义文档，它的行为非常清楚。