Flink SQL：在GROUP BY查询结果中重复分组键

我想在一个包含group by语句的表中在Flink SQL中进行简单查询。但是在结果中，group by语句中指定的列存在重复的行。那是因为我使用的是流媒体环境并且它不记得状态吗？

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
// configure Kafka consumer
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092"); // Broker default host:port
props.setProperty("group.id", "flink-consumer"); // Consumer group ID

FlinkKafkaConsumer011 flinkBlocksTransactionsConsumer = new FlinkKafkaConsumer011<>(args[0], new BlocksTransactionsSchema(), props);
flinkBlocksTransactionsConsumer.setStartFromEarliest();

DataStream blocksTransactions = env.addSource(flinkBlocksTransactionsConsumer);

tableEnv.registerDataStream("blocksTransactionsTable", blocksTransactions);

Table sqlResult

    = tableEnv.sqlQuery(
            "SELECT block_hash, count(tx_hash) " +
            "FROM blocksTransactionsTable " +
            "GROUP BY block_hash");

DataStream resultStream = tableEnv

    .toRetractStream(sqlResult, Row.class)
    .map(t -> {
        Row r = t.f1;
        String field2 = r.getField(0).toString();
        long count = Long.valueOf(r.getField(1).toString());
        return new Test(field2, count);
    })
    .returns(Test.class);

resultStream.print();

resultStream.addSink(new FlinkKafkaProducer011<>("localhost:9092", "TargetTopic", new TestSchema()));

env.execute();
我对block_hash列使用group by语句，但我有几次相同的block_hash。这是print（）的结果：

测试{FIELD2 = '0x2c4a021d514e4f8f0beb8f0ce711652304928528487dc7811d06fa77c375b5e1'，计数= 1}测试{FIELD2 = '0x2c4a021d514e4f8f0beb8f0ce711652304928528487dc7811d06fa77c375b5e1'，计数= 1}测试{FIELD2 = '0x2c4a021d514e4f8f0beb8f0ce711652304928528487dc7811d06fa77c375b5e1'，计数= 2}测试{FIELD2 = '0x780aadc08c294da46e174fa287172038bba7afacf2dff41fdf0f6def03906e60'，计数= 1}测试{ field2 ='0x182d31bd491527e1e93c4e44686057207ee90c6a8428308a2bd7b6a4d2e10e53'，count = 1}测试{field2 ='0x182d31bd491527e1e93c4e44686057207ee90c6a8428308a2bd7b6a4d2e10e53'，count = 1}

如何在不使用BatchEnvironment的情况下解决此问题？