How to integrate Beam SQL windowing query with KafkaIO?

First, we have a kafka input source in JSON format:

{"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}
... ...

What I'm trying to do is to aggregate the sum of the count of each word by every minute:

+---------+----------+---------------------+
| 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 |
+---------+----------+---------------------+

So this case would be perfectly fit the following Beam SQL statement:

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)

And below is my current working code using Beam's Java SDK to execute this streaming SQL query:

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();
}
}

My problem is that when I run this code and feed some messages into Kafka, there's no exception throwing out and it has received some messages from Kafka, but I cannot see it trigger the process for the windowing aggregation. And there's no result coming out as expected (like the table I shown before).

So does Beam SQL currently support the window syntax on unbounded Kafka input source? If it does, what's wrong with my current code? How can I debug and fix it? And is there any code example that integrates Beam SQL with KafkaIO?

Please help me! Thanks a lot!!

Looks like this was answered at https://lists.apache.org/thread.html/rea75c0eb665f90b8483e64bee96740ebb01942c606f065066c2ecc56%40%3Cuser.beam.apache.org%3E