Flink撤回机制不起作用

hi,all
我想简单的测试一下撤回机制，于是我写了以下代码
tableEnv.createTemporaryView("sensor", sensorTable);
val resultSqlTable = tableEnv.sqlQuery("select country, count(order_id) as cnt from sensor group by country");
tableEnv.toRetractStream[Wate
rSensorCnt]
(resultSqlTable).print("result");

然后在socket发送以下数据：
001 usa
002 usa
003 china
002 china
004 usa

我预期在控制台得到的结果应该是
usa, 2
china, 2

但是结果却是：
usa, 3
china, 2

本应该usa撤回一条才符合我对撤回机制的理解，但是usa并没有减少？
大家可以帮我消除疑惑吗？
如果您能在百忙之中抽空解答，我将非常感激！*来自志愿者整理的flink邮件归档