flink如何处理两个流按键连接，但没有匹配的键

public class KeyedStreamJoinExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建第一个输入流
        DataStream<Tuple2<String, Integer>> input1 = env.fromElements(
                new Tuple2<>("key1", 1),
                new Tuple2<>("key2", 2),
                new Tuple2<>("key3", 3)
        );

        // 创建第二个输入流
        DataStream<Tuple2<String, String>> input2 = env.fromElements(
                new Tuple2<>("key1", "value1"),
                new Tuple2<>("key2", "value2"),
                new Tuple2<>("key4", "value4")
        );

        // 将两个输入流连接起来，并使用CoProcessFunction处理
        DataStream<String> result = input1
                .keyBy(value -> value.f0)
                .connect(input2.keyBy(value -> value.f0))
                .process(new KeyedStreamJoinFunction());

        result.print();

        env.execute("Keyed Stream Join Example");
    }

    public static class KeyedStreamJoinFunction extends CoProcessFunction<
            Tuple2<String, Integer>,
            Tuple2<String, String>,
            String
            > {
        private ValueState<Integer> input1State;
        private ValueState<String> input2State;

        @Override
        public void open(Configuration parameters) throws Exception {
            // 初始化状态
            input1State = getRuntimeContext().getState(new ValueStateDescriptor<>("input1State", Integer.class));
            input2State = getRuntimeContext().getState(new ValueStateDescriptor<>("input2State", String.class));
        }

        @Override
        public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {
            // 处理第一个输入流的元素
            input1State.update(value.f1);
            ctx.timerService().registerProcessingTimeTimer(System.currentTimeMillis() + 1000);
        }

        @Override
        public void processElement2(Tuple2<String, String> value, Context ctx, Collector<String> out) throws Exception {
            // 处理第二个输入流的元素
            input2State.update(value.f1);
            ctx.timerService().registerProcessingTimeTimer(System.currentTimeMillis() + 1000);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            // 定时器触发时处理没有匹配的键
            if (input1State.value() == null) {
                out.collect("No match for key in input1: " + ctx.getCurrentKey());
            }
            if (input2State.value() == null) {
                out.collect("No match for key in input2: " + ctx.getCurrentKey());
            }
        }
    }
}