我正在为一个个人项目试验Apache Flink，我一直在努力将结果流输出到StdOut并将其发送到Kafka主题orders-output。
我的目标是在3分钟的滚动时间窗口内计算每个产品的价格字段的总和。基本上，Apache flink作业，从两个Kafka源流（orders-a和orders-b）接收JSON格式的字符串订单，将它们连接在一起，获得形状为(product_name, product_price (double)的元组，然后按产品对其进行分组，应用3分钟的滚动窗口，并使用ReduceFunction计算该窗口中每个产品的价格总和。代码如下：

FlinkKafkaConsumer<String> consumerA = new FlinkKafkaConsumer<>("orders-a", new SimpleStringSchema(), props);
    FlinkKafkaConsumer<String> consumerB = new FlinkKafkaConsumer<>("orders-b", new SimpleStringSchema(), props);

    DataStream<String> streamA = env.addSource(consumerA);
    DataStream<String> streamB = env.addSource(consumerB);

    DataStream<Tuple2<String,Double>> mergedOrders = streamA
            .union(streamB)
            .map(new MapFunction<String, Tuple2<String, Double>>() {
                @Override
                public Tuple2<String, Double> map(String s) throws Exception {
                    return DataHelper.getTuple(s);
                }
            });

    DataStream<Tuple2<String, Double>> totals = mergedOrders
            .keyBy(value -> value.f0)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(new ReduceFunction<Tuple2<String, Double>>() {
                public Tuple2<String, Double> reduce(Tuple2<String, Double> v1, Tuple2<String, Double> v2) {
                    return new Tuple2<>(v1.f0, v1.f1 + v2.f1);
                }
            });
    DataStream<String> result = totals.map(new MapFunction<Tuple2<String, Double>, String>() {
        @Override
        public String map(Tuple2<String, Double> stringDoubleTuple2) throws Exception {
            LOG.info(stringDoubleTuple2.toString());
            return stringDoubleTuple2.toString();
        }
    });

    result.print();
    result.addSink(new FlinkKafkaProducer<>("orders-output", new SimpleStringSchema(), props));

    env.execute("Product revenues per 3m");

这段代码完成了前面所描述的工作，DataHelper只是一个自定义助手类，我创建它来帮助我将接收到的订单从JSON字符串转换为Tuple2和其他类型。作业启动后运行良好（我在本地运行所有内容），我甚至可以看到数据在Flink UI中接收（见下图）。

问题是，我在StdOut（在终端和Flink UI中）和Kafka输出主题中都没有看到任何结果（我在另一个终端中独立启动了orders-output的消费者，但我没有收到任何结果）。
我会很感激你的帮助，因为我已经被困在这两天了。