[BUG]Consumer will be blocked after the server throws multiple exceptions.

[deemogsw]

Describe： RocketMQ server has some problems in consumer group verification.Source will receive exception when it pulls record from broker. After five retries ,none reocrds would be pulled and job is still in running state. Analyse: RocketMQSourceFunction pulls record from RocketMQ in a thread. When an exception occurs, it can be retried up to five times.Then thread of consumer would finish,but main thread still be blocked by awaitTermination function.So the function of run never ends and subTask can't exit normally.

After server of rocketmq throws multiple exceptions. Even if all threads of rmq-pull had finished ,the job is still in running state.This is a big risk in the production.

[deemogsw]

An exception can't be throwed in rmq-pull-thread,So function of run can't catch exception and flink still believes that the program has no exceptions.