Watcher leak in Zookeeper HA mode

We observe a watcher leak in our OLAP stress test when enabling Zookeeper HA mode. TM's watches on the leader of JobMaster has not been stopped after job finished.

Here is how we re-produce this issue:

• Start a session cluster and enable Zookeeper HA mode.
• Continuously and concurrently submit short queries, e.g. WordCount to the cluster.
• echo -n wchp | nc {zk host} {zk port} to get current watches.

We can see a lot of watches on /flink/{cluster_name}/leader/{job_id}/connection_info.