flink作业延迟监控

[casel.chen]

想实现flink sql作业延迟监控，例如flink 
sql作业将kafka数据写入mysql，记kafka记录中的事件时间为T0，发到kafka时间是T1，写入mysql的时间为T2，现要统计如下时间差（延迟）
1. T2 - T1 ：flink sql作业延迟
2. T2 - T0 ：端到端延迟，包括flink sql作业延迟和数据写入kafka延迟
请问:
1) 要如何暴露这2个时间差作为metrics？
2) 中间算子的处理时长能暴露吗？