KStream和KTable之间的时间语义

Question

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

>

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

在我的拓扑中，第一步是从这两个“表”主题创建“干净的”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);
}

请注意，我显式地分配记录时间，因为表行将在它们最初发布后被CDC'ed“年”。该函数目前正在做的是伪造从2010-01-01开始的时间，并使用AtomicInteger为每个消耗的实体添加1毫秒。它对表A这样做，但对DA不这样做（我将在后面解释原因）。

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 ] ));
}

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] ]
}

因此，我没有所有的“加入点击”，我所期待的，这也是不确定的。基于这句话，我的理解是KTable与GlobalKTable和leftJoin（）与outerJoin（）之间有什么区别？却是相反的：

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

我的经验是，到目前为止，这还没有发生。我还可以很容易地看到我的应用程序在消耗了Table_DA流中的所有条目（它碰巧是Table_DA流的10倍小）之后是如何继续在Table_A主题中翻腾的。

Answer 1

时间戳同步是2.1.0版本之前的最大努力（参见https://issues.apache.org/jira/browse/kafka-3514)。

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