[BUG] topic route changed,job can't recover normally

1.Add partition
Flink will check partition is changed in function of snapshotState.If route changed,job will switch state from running to failed.
Then job will recover from checkpoint without new message queue record.Offset table alse has none new message queue,NPE will throw in RockertMQSourceFunction$run

2.Decrease partition
If partition decreased,can't connect execption will throw in RockertMQSourceFunction$run.After five reries,function of run will close normally.Job will switch state from running to finished rather than failed because thread poll will swallow the connection exception

Start by tracing snapshotState and RockertMQSourceFunction$run, then reproduce both partition-increase and partition-decrease cases described in the issue. Check how missing offset-table entries and connection exceptions affect recovery and job state; done means neither case produces an NPE or is incorrectly reported as a successful finish.

Written by the indexing model from the issue text.