背景：我们使用flink来使用kafka的数据，并在小时分区中写入hdfs。我们不设置水印，因为我们希望所有的记录都被消耗掉。所以我们得到的目录是

/data/app1/day=2019-01-01/hour=01/
/data/app1/day=2019-01-01/hour=02/
...
/data/app1/day=2019-01-01/hour=23/

其他的工作需要我们的数据作为输入，所以如果一些迟来的数据可能会在几个小时后到达，那么这些数据是不完整的，这会导致下游工作的不正确结果。
所以我的问题是如何保证或检测flink中的数据完整性？