Auron Flink native operator support watermark

We've added support for Flink's Watermark mechanism in the Kafka Source. As more and more operators get native support, we plan to merge connected native operators together for execution to reduce the R2C (row-to-columnar) conversion overhead. However, Auron's current Watermark mechanism isn't designed for this scenario, so we need to design a solution that fits this use case while still satisfying the Watermark constraints.