JdbcUpsertTableSink输出的时候报‘Duplicate entry’问题

如题，按照官方文档，当mysql表定义了primary key的时候，会使用UpsertTableSink，并且会使用insert on duplicate方式写入。

但我在使用中，发现报了 duplicate entry的错误。例如： Caused by: com.mysql.jdbc.exceptions.jdbc4. MySQLIntegrityConstraintViolationException: Duplicate entry '2036-feed_landing_box_news-2000-202011231405' for key 'uniq_ssmt' at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at sun.reflect.NativeConstructorAccessorImpl.newInstance( NativeConstructorAccessorImpl.java:62) at sun.reflect.DelegatingConstructorAccessorImpl.newInstance( DelegatingConstructorAccessorImpl.java:45) at java.lang.reflect.Constructor.newInstance(Constructor.java:423) at com.mysql.jdbc.Util.handleNewInstance(Util.java:411) at com.mysql.jdbc.Util.getInstance(Util.java:386) at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:1041) at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:4190) at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:4122) at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2570) at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2731) at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2818) at com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement .java:2157) at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement .java:2460) at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement .java:2377) at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement .java:2361) at com.mysql.jdbc.PreparedStatement.executeBatchedInserts( PreparedStatement.java:1793)

（2） 此外，还有个小奇怪点，202011231405的数据，其他该时间的数据都在06分输出了（我设置了1min的maxOutOfOrder）。 但这个冲突的entry是在14.11分那一波才报错的。*来自志愿者整理的flink邮件归档