在使用Kafka Streams的WindowStateStore进行聚合处理时，处理未来时间戳可能存在的危险陷阱

2 年 ago

新, 韵

2 minutes

Kafka Streams有一个名为Window Aggregation的机制，可以对事件进行计数等操作，以特定的时间间隔为单位。

所以，我写了一个利用它来计算每个用户事件执行次数的处理方法，但奇怪的是结果却不符合预期。
在测试代码中无法复制，并且出现了概率性结果偏差的情况。

我非常沉迷于其中，但问题出在处理未来时间戳的原因上。

验证版本

kafka-streams-2.4.0

問題的詳細

尽管数据已经成功写入并传递到后端的Kafka主题，但当尝试重新获取数据时，无法获取到现有的聚合计数。
因此计数被重置，并且其他地方也无法获取结果的状态。

紐付いているKafkaのトピックには確実にデータが届いている

原因：RocksDB的Segment因未来时间戳的影响而过期。

Kafka Streams的WindowStore中存在着保留期限。超过保留期限的数据将会被删除。
以以下例子为准，保持期限为30天。

  private StoreBuilder<WindowStore<String, Counter>> counterStore =
      Stores.windowStoreBuilder(
              Stores.persistentWindowStore(
                  "counter",
                  Duration.ofDays(30),
                  Duration.ofHours(24),
                  false),
              Serdes.String(),
              Serdes.Integer())
          .withCachingEnabled();

另外，WindowStore在内部通过Segment这个单位对数据目录进行划分。
Segment的间隔是保留期的一半时间间隔。（最短为1分钟）

大致上表现出以下的动作。

timestampとsegment intervalからsegment idを算出する

    @Override
    public String segmentName(final long segmentId) {
        // (1) previous format used - as a separator so if this changes in the future
        // then we should use something different.
        // (2) previous format used : as a separator (which did break KafkaStreams on Windows OS)
        // so if this changes in the future then we should use something different.
        return name + "." + segmentId * segmentInterval;
    }

    @Override
    public long segmentId(final long timestamp) {
        return timestamp / segmentInterval;
    }

対象のSegmentを特定しそこにデータをputする

    @Override
    public void put(final Bytes key,
                    final byte[] value) {
        final long timestamp = keySchema.segmentTimestamp(key);
        observedStreamTime = Math.max(observedStreamTime, timestamp);
        final long segmentId = segments.segmentId(timestamp);
        final S segment = segments.getOrCreateSegmentIfLive(segmentId, context, observedStreamTime);
        if (segment == null) {
            expiredRecordSensor.record();
            LOG.debug("Skipping record for expired segment.");
        } else {
            segment.put(key, value);
        }
    }

cleanup対象のsegmentは削除フラグを付与されてcloseされる。

    @Override
    public S getOrCreateSegmentIfLive(final long segmentId,
                                      final InternalProcessorContext context,
                                      final long streamTime) {
        final long minLiveTimestamp = streamTime - retentionPeriod;
        final long minLiveSegment = segmentId(minLiveTimestamp);

        final S toReturn;
        if (segmentId >= minLiveSegment) {
            // The segment is live. get it, ensure it's open, and return it.
            toReturn = getOrCreateSegment(segmentId, context);
        } else {
            toReturn = null;
        }

        cleanupEarlierThan(minLiveSegment);
        return toReturn;
    }

    private void cleanupEarlierThan(final long minLiveSegment) {
        final Iterator<Map.Entry<Long, S>> toRemove =
            segments.headMap(minLiveSegment, false).entrySet().iterator();

        while (toRemove.hasNext()) {
            final Map.Entry<Long, S> next = toRemove.next();
            toRemove.remove();
            final S segment = next.getValue();
            segment.close();
            try {
                segment.destroy();
            } catch (final IOException e) {
                log.error("Error destroying {}", segment, e);
            }
        }
    }

如果这里收到了一条具有相当未来时间戳的记录，observedStreamTime将被固定在未来的时间点上，这样在从中减去保留期之后，过去的StateStore将全部关闭且无法使用！

结论 –

简而言之，如果一次写入超过保留期限的未来时间戳，那么就无法获取或写入比该时间戳更早的结果。即使随后出现了正确的当前时间戳记录，也无法使用它。

如果出现具有一个月以上未来时间戳的记录，并且这些记录使用相同的分区，那么与这些记录关联的计数将会被破坏。而且，在发生这种情况时，没有任何错误日志或警告信息可供参考。目前，如果无法信任正在进行聚合的时间戳偏移量，那么必须始终排除具有超过一定未来时间戳的记录。即使你信任时间戳，也最好做好防护措施。

我非常担心诚实和大量的努力，以及对整个功能的可靠性非常不满意，但好在终于找到了问题的原因。

由于Kafka Streams的处理涉及到时间戳的机制，因此对于可能出现严重时间偏差数据的系统，需要注意处理的方法。