首先，我们有一个json格式的kafka输入源：

{"event_time": "2020-08-23 18:36:10", "word": "apple", "cnt": 1}
{"event_time": "2020-08-23 18:36:20", "word": "banana", "cnt": 1}
{"event_time": "2020-08-23 18:37:30", "word": "apple", "cnt": 2}
{"event_time": "2020-08-23 18:37:40", "word": "apple", "cnt": 1}
... ...

我要做的是把每分钟每一个单词的总数加起来：

+---------+----------+---------------------+
| word    | SUM(cnt) |   window_start      |
+---------+----------+---------------------+
| apple   |    1     | 2020-08-23 18:36:00 |
+---------+----------+---------------------+
| banana  |    1     | 2020-08-23 18:36:00 |
+---------+----------+---------------------+
| apple   |    3     | 2020-08-23 18:37:00 |
+---------+----------+---------------------+

所以这个例子非常适合下面的beam sql语句：

SELECT word, SUM(cnt), TUMBLE_START(event_time, INTERVAL '1' MINUTE) as window_start
FROM t_count_stats
GROUP BY word, TUMBLE(event_time, INTERVAL '1' MINUTE)

下面是我当前使用beam的java sdk执行此流式sql查询的工作代码：

import avro.shaded.com.google.common.collect.ImmutableMap;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.extensions.sql.SqlTransform;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;

import java.util.ArrayList;
import java.util.List;

public class KafkaBeamSqlTest {

    private static DateTimeFormatter dtf = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");

    public static void main(String[] args) {
        // create pipeline
        PipelineOptions kafkaOption = PipelineOptionsFactory.fromArgs(args)
                .withoutStrictParsing()
                .as(PipelineOptions.class);
        Pipeline pipeline = Pipeline.create(kafkaOption);

        // create kafka IO
        KafkaIO.Read<String, String> kafkaRead =
                KafkaIO.<String, String>read()
                        .withBootstrapServers("127.0.0.1:9092")
                        .withTopic("beamKafkaTest")
                        .withConsumerConfigUpdates(ImmutableMap.of("group.id", "client-1"))
                        .withReadCommitted()
                        .withKeyDeserializer(StringDeserializer.class)
                        .withValueDeserializer(StringDeserializer.class)
                        .commitOffsetsInFinalize();

        // read from kafka
        PCollection<KV<String, String>> messages = pipeline.apply(kafkaRead.withoutMetadata());

        // build input schema
        Schema inputSchema = Schema.builder()
                .addStringField("word")
                .addDateTimeField("event_time")
                .addInt32Field("cnt")
                .build();

        // convert kafka message to Row
        PCollection<Row> rows = messages.apply(ParDo.of(new DoFn<KV<String, String>, Row>(){
            @ProcessElement
            public void processElement(ProcessContext c) {
                String jsonData = c.element().getValue();

                // parse json
                JSONObject jsonObject = JSON.parseObject(jsonData);

                // build row
                List<Object> list = new ArrayList<>();
                list.add(jsonObject.get("word"));
                list.add(dtf.parseDateTime((String) jsonObject.get("event_time")));
                list.add(jsonObject.get("cnt"));
                Row row = Row.withSchema(inputSchema)
                        .addValues(list)
                        .build();

                System.out.println(row);

                // emit row
                c.output(row);
            }
        }))
                .setRowSchema(inputSchema);

        // sql query
        PCollection<Row> result = PCollectionTuple.of(new TupleTag<>("t_count_stats"), rows)
                .apply(
                        SqlTransform.query(
                                "SELECT word, SUM(cnt), TUMBLE_START(event_time, INTERVAL '1' MINUTE) as window_start\n" +
                                "FROM t_count_stats\n" +
                                "GROUP BY word, TUMBLE(event_time, INTERVAL '1' MINUTE)"
                        )
                );

        // sink results back to another kafka topic
        result.apply(MapElements.via(new SimpleFunction<Row, KV<String, String>>() {
            @Override
            public KV<String, String> apply(Row input) {

                System.out.println("result: " + input.getValues());

                return KV.of(input.getValue("word"), "result=" + input.getValues());
            }
        }))
                .apply(KafkaIO.<String, String>write()
                        .withBootstrapServers("127.0.0.1:9092")
                        .withTopic("beamPrint")
                        .withKeySerializer(StringSerializer.class)
                        .withValueSerializer(StringSerializer.class));

        // run
        pipeline.run();
    }
}

我的问题是，当我运行这段代码并将一些消息输入kafka时，没有异常抛出，它已经从kafka接收到一些消息，但是我看不到它触发窗口聚合的进程。结果并没有像预期的那样（就像我之前展示的表格一样）。
那么beamsql目前是否支持无界kafka输入源的窗口语法呢？如果是的话，我现在的代码怎么了？如何调试和修复它？有没有任何将beamsql与kafkaio集成在一起的代码示例？
请帮帮我！谢谢！！