KStream 到 KStream 连接 - 如果窗口中没有匹配的记录，输出记录将发布可配置的时间

Question

需要一些关于 KStream/KTable 用法用例的意见/帮助。

场景：

我有两个具有公共关键字requestId的主题。

1. input_time启动时间
2. completion_time（Request Id， EndTime）

input_time中的数据在时间t1填充，completion_time中的数据在时间tn填充（n是进程完成所需的时间）。

目的通过连接来自主题的数据来比较请求所用的时间，并在违反阈值时间的情况下发出警报。

可能会发生这样的情况：该过程可能会失败，并且数据可能根本没有到达请求completion_time主题。在这种情况下，我们打算使用检查，如果当前时间自开始时间以来远远超过特定（假设5s）阈值。

1. input_time（要求1，100） completion_time（要求1，104） --

已尝试的选项。1）尝试了KTable-KTable和KStream-Kstream外连接，但第三种情况总是失败。

    final KTable<String,Long> startTimeTable =   builder.table("input_time",Consumed.with(Serdes.String(),Serdes.Long()));
    final KTable<String,Long> completionTimeTable = builder.table("completion_time",Consumed.with(Serdes.String(),Serdes.Long()));     
    KTable<String,Long> thresholdBreached =startTimeTable .outerJoin(completionTimeTable,
            new MyValueJoiner());
    thresholdBreached.toStream().filter((k,v)->v!=null)
            .to("finalTopic",Produced.with(Serdes.String(),Serdes.Long()));

木匠

 public Long apply(Long startTime,Long endTime){

        // if input record itself is not available then we cant use any alerting.
        if (null==startTime){
            log.info("AlertValueJoiner check: the start time itself is null so returning null");
            return null;
        }
        // current processing time is the time used.
        long currentTime= System.currentTimeMillis();
        log.info("Checking startTime {} end time {} sysTime {}",startTime,endTime,currentTime);
        if(null==endTime && currentTime-startTime>5000){
            log.info("Alert:No corresponding record from file completion yet currentTime {} startTime {}"
                    ,currentTime,startTime);
            return currentTime-startTime;
        }else if(null !=endTime && endTime-startTime>5000){
            log.info("Alert: threshold breach for file completion startTime {} endTime {}"
                    ,startTime,endTime);
            return endTime-startTime;
        }
    return null;
    }

2） 尝试了根据线程推荐的自定义逻辑方法 如何管理 Kafka KStream 到 Kstream 窗口式联接？-- 此方法已停止适用于方案 2 和 3。

是否有任何使用 DSL 或处理器处理所有三种情况的情况？

不确定我们是否可以使用某种标点符号来监听窗口何时更改并检查当前窗口中的流记录，如果没有找到匹配的记录，则使用systime生成结果。

Answer 1

由于逻辑的性质，它肯定必须通过DSL和处理器API的组合来完成。

1. 使用自定义转换器和状态存储与配置的值进行比较。（情况1