KStream 和 KTable 之间的时间语义

Question

我正在尝试构建以下拓扑：

1. 使用 Debezium 连接器，我拉了 2 个表（我们称它们为表 A 和 DA）。根据 DBZ，存储表行的主题具有结构 { before: "...", after: "..." }。

2. 我的拓扑的第一步是从这两个“表”主题创建“干净的”KStreams。那里的子拓扑大致如下所示：

private static KStream<String, TABLE_A.Value> getTableARowByIdStream(
    StreamsBuilder builder, Properties streamsConfig) {
  return builder
      .stream("TABLE_A", Consumed.withTimestampExtractor(Application::getRowDate))
      .filter((key, envelope) -> [ some filtering condition ] )
      .map((key, envelope) -> [ maps to TABLE_A.Value ] )
      .through(tableRowByIdTopicName);
}

3. 请注意，我明确分配了记录时间，因为表行在最初发布后将被 CDC 标记为“年”。该函数目前正在做的是伪造从 2010-01-01 开始的时间，并使用AtomicInteger为每个消耗的实体添加 1 毫秒。它对表 A 执行此操作，但对 DA 不执行此操作（稍后我将解释原因）。

4. 拓扑的第 2 阶段是基于表 A 的“已清理”主题构建 1 个 KTable，如下所示：

private static KTable<String, EntityInfoList> getEntityInfoListById(
    KStream<String, TABLE_A.Value> tableAByIdStream) {
  return tableAByIdStream
      .map((key, value) -> [ some mapping ] )
      .groupByKey()
      .aggregate(() -> [ builds up a EntityInfoList object ] ));
}

5. 最后，在准备好 KTable 后，我将通过 DA 与 KStream 一起加入他们，如下所示：

private static KStream<String, OutputTopicEntity> getOutputTopicEntityStream(
    KStream<String, Table_DA.Value> tableDAStream,
    KTable<String, EntityInfoList> tableA_KTable) {

  KStream<String, Table_DA>[] branches = tableDAStream.branch(
      (key, value) -> [ some logic ],
      (key, value) -> true);

  KStream<String, OutputTopicEntity> internalAccountRefStream = branches[0]
      .join(
          tableA_KTable,
          (streamValue, tableValue) -> [ some logic to build a list of OutputTopicEntity ])
      .flatMap((key, listValue) -> [ some logic to flatten it ]));

   [ similar logic with branch[1] ]
}

我的问题是，尽管我“伪造”了来自 ​​Table_A 主题的记录的时间（我已经验证它们使用 kafkacat 引用了 2010/01/01）和 Table_DA 中的条目（连接的流端） ) 有今天 '2019/08/14' 左右的时间戳)，在将 Table_A 中的所有记录摄取到 KTable 之前，Kafka Streams 似乎不会从 Table_DA KStream 中读取任何条目。

因此，我没有预期的所有“加入命中”，而且它也是不确定的。我基于这句话的理解KTable vs GlobalKTable 和 leftJoin() vs outerJoin() 有什么区别？正好相反：

对于流表连接，Kafka Stream 对齐基于记录时间戳排序的记录处理。因此，对表的更新与您的流记录对齐。

到目前为止，我的经验是这不会发生。我还可以很容易地看到我的应用程序在消耗了 Table_DA 流中的所有条目后如何继续通过 Table_A 主题方式搅动（它恰好小了 10 倍）。

难道我做错了什么？

Answer 1

在 2.1.0 版本发布之前，最好进行时间戳同步（参见https://issues.apache.org/jira/browse/KAFKA-3514）。

从 2.1.0 开始，时间戳严格同步。但是，如果一个输入没有任何数据，Kafka Streams 将按照KIP-353中的描述“强制”处理，以避免永远阻塞。如果您有突发输入，并且希望在一个输入没有数据的情况下“阻止”处理一段时间，您可以通过 KIP-353增加配置参数max.task.idle.ms（默认为），如 2.1.0 中所述。0