Kafka Streams - JoinWindows 类型的 (Duration) 方法已弃用答案

【问题标题】：Kafka Streams - The method of(Duration) from the type JoinWindows is deprecatedKafka Streams - JoinWindows 类型的 (Duration) 方法已弃用
【发布时间】：2022-12-07 04:49:43
【问题描述】：

我想使用 JoinWindows.of(Duration) 加入 KStream<String, String> 和 KTable<Windowed, int[]> 以获取过去一小时的结果。

我的代码如下：

Duration windowSize = Duration.ofMinutes(60);
Duration advanceSize = Duration.ofMinutes(1);
TimeWindows hoppingWindow = TimeWindows.ofSizeWithNoGrace(windowSize).advanceBy(advanceSize);
Duration joinWindowSizeMs = Duration.ofHours(1);

// Aggregate to get [sum, count] in the last time window
KTable<Windowed<String>, int[]> averageTemp = mainStreamStandard.groupByKey()
.windowedBy(hoppingWindow)
.aggregate( () -> new int[]{0 ,0}, (aggKey, newVal, aggValue) -> {
        aggValue[0] += Integer.valueOf(newVal.split(":")[1]);
        aggValue[1] += 1;  
        return aggValue;
        }, Materialized.with(Serdes.String(), new IntArraySerde()));
        
// Join weather stations with their [sum,count] and their respective red alert events
KStream<String, String> joined = mainStreamAlert.join(averageTemp,
JoinWindows.of(joinWindowSizeMs),
(leftValue, rightValue) -> "left/" + leftValue + "/right/" + rightValue[0]/rightValue[1]);

它给出了一个错误，说“JoinWindows 类型的（Duration）方法已被弃用”。它还告诉我将“join”方法更改为“leftJoin”，但它没有任何改变。

有什么更好的方法来做到这一点？

【问题讨论】：

标签： java join apache-kafka time apache-kafka-streams

【解决方案1】：

of(Duration timeDifference) 确实已弃用。

这个方法的直接替代是ofTimeDifferenceAndGrace(Duration timeDifference, Duration afterWindowEnd)。

已弃用的方法有 24 小时的默认宽限期。所以你的代码看起来像这样：

...
Duration joinWindowSizeMs = Duration.ofHours(1);
Duration gracePeriod = Duration.ofHours(24);
...
// Join weather stations with their [sum,count] and their respective red alert events
KStream<String, String> joined = mainStreamAlert.join(averageTemp,
JoinWindows.ofTimeDifferenceAndGrace(joinWindowSizeMs, gracePeriod),
(leftValue, rightValue) -> "left/" + leftValue + "/right/" + rightValue[0]/rightValue[1]);

JoinWindows documentation

【讨论】：