kafka-streams：如果缺少单个元素，则重试完成消息连接

Question

{
  "order_id": 123456789,
  "user_id": 987654,
  "placed_at": "2020-07-20T11:31:00",
  "amount": 5.79,
  "items" : [
     {"item_id": 13579, "quantity": 1, "price": 1.23},
     {"item_id": 24680, "quantity": 1, "price": 4.56}
  ]
}

我当前的方法是从placements获取传入消息，将其拆分为N个消息（其中N是item_id数组的长度），对item_id执行左联接以添加项目描述，然后将结果流分组到order_id（丰富的拆分消息的关键字）上，以重构完整消息。

问题是描述可能会延迟几秒钟到达，所以在一些罕见的情况下，我得到的重构消息的条目比原始未充实的少。

我已经在自定义联接示例中看到了这种方法。对我的案子来说挺好的，可惜不完全适合。的确，在我的案例中，如果缺少单个项目的描述，那么完整的消息应该被延迟。目前我不知道在这种情况下该如何进行。欢迎任何建议。

Answer 1

在仔细分析了自定义联接示例之后，解决方案是稍微改变一下它的逻辑。

下面是示例的节选：

private static final class StreamTableJoinStreamSideLogic
      implements TransformerSupplier<String, Double, KeyValue<String, Pair<Double, Long>>> {

/* ... */

private KeyValue<String, Pair<Double, Long>> sendFullJoinRecordOrWaitForTableSide(final String key,
                                                                                          final Double value,
                                                                                          final long streamRecordTimestamp) {
          final ValueAndTimestamp<Long> tableValue = tableStore.get(key);
          if (tableValue != null &&
              withinAcceptableBounds(Instant.ofEpochMilli(tableValue.timestamp()), Instant.ofEpochMilli(streamRecordTimestamp))) {
            final KeyValue<String, Pair<Double, Long>> joinRecord = KeyValue.pair(key, new Pair<>(value, tableValue.value()));
            LOG.info("Table data available for key {}, sending fully populated join message {}", key, joinRecord);
            return joinRecord;
          } else {
            LOG.info("Table data unavailable for key {}, sending the join result as null", key);
            return KeyValue.pair(key, new Pair<>(value, null));
          }
        }

/* ... */

}

特别是，需要修改SendFullJoinRecordorWaitForTableSide()方法，以便以全或全的方式将相同的逻辑应用于Items。