flink kafka consumer support sync-savepoint

The current Auron Flink Kafka consumer is implemented using Flink's legacy source. Periodic checkpoints and non-sync savepoints work fine, but Flink's sync-savepoint flow goes: cancel first, then trigger the snapshot logic. Since the current Auron implementation uses a while infinite loop, it can't detect the cancel event, which causes the sync-savepoint to hang.