[rocketmq-runtime] stop Replicator error

启动一个 Replicator后，会有多个task
停止该connector，task会报错
1. 调用 http://localhost:8081/getAllocatedInfo 获取如下
```
working connectors:
connectorName:replicator-test
Configs:{"properties":{"source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter","replicator-store-topic":"replicatorTopic","task-parallelism":"2","taskDivideStrategy":"0","white-list":"fileTopic","source-cluster":"RocketMQCluster-172","connector-class":"org.apache.rocketmq.replicator.RmqSourceReplicator","target-rocketmq":"47.106.210.217:9876","source-rocketmq":"47.106.219.172:9876","target-cluster":"RocketMQCluster-217","update-timestamp":"1577025620944"}}
working tasks:
connectorName:replicator-test
Configs:{"properties":{"source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter","sourceRocketmq":"47.106.219.172:9876","task-class":"org.apache.rocketmq.replicator.RmqSourceTask","dataType":"0","storeTopic":"replicatorTopic","taskTopicList":"[{\"brokerName\":\"broker-b\",\"queueId\":7,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":5,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":3,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":1,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"}]","update-timestamp":"1577025620944"}}
connectorName:replicator-test
Configs:{"properties":{"source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter","sourceRocketmq":"47.106.219.172:9876","task-class":"org.apache.rocketmq.replicator.RmqSourceTask","dataType":"0","storeTopic":"replicatorTopic","taskTopicList":"[{\"brokerName\":\"broker-b\",\"queueId\":6,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":4,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":2,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":0,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"}]","update-timestamp":"1577025620944"}}
```
停止该connector报错

```
2019-12-22 22:49:00 INFO pool-4-thread-3 - Source task stop, config:{"properties":{"source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter","sourceRocketmq":"47.106.219.172:9876","task-class":"org.apache.rocketmq.replicator.RmqSourceTask","dataType":"0","storeTopic":"replicatorTopic","taskTopicList":"[{\"brokerName\":\"broker-b\",\"queueId\":7,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":5,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":3,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":1,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"}]","update-timestamp":"1577025620944"}}
2019-12-22 22:49:00 INFO RebalanceService - Source task stop, connector name replicator-test, config ConnectKeyValue{properties={source-record-converter=org.apache.rocketmq.connect.runtime.converter.JsonConverter, sourceRocketmq=47.106.219.172:9876, task-class=org.apache.rocketmq.replicator.RmqSourceTask, dataType=0, storeTopic=replicatorTopic, taskTopicList=[{"brokerName":"broker-b","queueId":7,"targetTopic":"fileTopic","topic":"fileTopic"},{"brokerName":"broker-b","queueId":5,"targetTopic":"fileTopic","topic":"fileTopic"},{"brokerName":"broker-b","queueId":3,"targetTopic":"fileTopic","topic":"fileTopic"},{"brokerName":"broker-b","queueId":1,"targetTopic":"fileTopic","topic":"fileTopic"}], update-timestamp=1577025620944}}
2019-12-22 22:49:00 INFO RebalanceService - Source task stop, connector name replicator-test, config ConnectKeyValue{properties={source-record-converter=org.apache.rocketmq.connect.runtime.converter.JsonConverter, sourceRocketmq=47.106.219.172:9876, task-class=org.apache.rocketmq.replicator.RmqSourceTask, dataType=0, storeTopic=replicatorTopic, taskTopicList=[{"brokerName":"broker-b","queueId":6,"targetTopic":"fileTopic","topic":"fileTopic"},{"brokerName":"broker-b","queueId":4,"targetTopic":"fileTopic","topic":"fileTopic"},{"brokerName":"broker-b","queueId":2,"targetTopic":"fileTopic","topic":"fileTopic"},{"brokerName":"broker-b","queueId":0,"targetTopic":"fileTopic","topic":"fileTopic"}], update-timestamp=1577025620944}}
2019-12-22 22:49:00 ERROR pool-4-thread-4 - Rocketmq replicator task poll error, current config: {"dataType":0,"nextPosition":57,"sourceRocketmq":"47.106.219.172:9876","storeTopic":"replicatorTopic","taskTopicList":"[{\"brokerName\":\"broker-b\",\"queueId\":6,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":4,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":2,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":0,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"}]"}
org.apache.rocketmq.client.exception.MQClientException: The consumer is not in running status, SHUTDOWN_ALREADY
See http://rocketmq.apache.org/docs/faq/ for further details.
at org.apache.rocketmq.client.impl.consumer.DefaultMQPullConsumerImpl.isRunning(DefaultMQPullConsumerImpl.java:105) ~[rocketmq-replicator-0.1.0-SNAPSHOT-jar-with-dependencies.jar:4.5.2]
at org.apache.rocketmq.client.impl.consumer.DefaultMQPullConsumerImpl.pullSyncImpl(DefaultMQPullConsumerImpl.java:228) ~[rocketmq-replicator-0.1.0-SNAPSHOT-jar-with-dependencies.jar:4.5.2]
at org.apache.rocketmq.client.impl.consumer.DefaultMQPullConsumerImpl.pull(DefaultMQPullConsumerImpl.java:181) ~[rocketmq-replicator-0.1.0-SNAPSHOT-jar-with-dependencies.jar:4.5.2]
at org.apache.rocketmq.client.impl.consumer.DefaultMQPullConsumerImpl.pull(DefaultMQPullConsumerImpl.java:175) ~[rocketmq-replicator-0.1.0-SNAPSHOT-jar-with-dependencies.jar:4.5.2]
at org.apache.rocketmq.client.consumer.DefaultMQPullConsumer.pull(DefaultMQPullConsumer.java:313) ~[rocketmq-replicator-0.1.0-SNAPSHOT-jar-with-dependencies.jar:4.5.2]
at org.apache.rocketmq.replicator.RmqSourceTask.pollCommonMessage(RmqSourceTask.java:148) [rocketmq-replicator-0.1.0-SNAPSHOT-jar-with-dependencies.jar:na]
at org.apache.rocketmq.replicator.RmqSourceTask.poll(RmqSourceTask.java:73) [rocketmq-replicator-0.1.0-SNAPSHOT-jar-with-dependencies.jar:na]
at org.apache.rocketmq.connect.runtime.connectorwrapper.WorkerSourceTask.run(WorkerSourceTask.java:140) [rocketmq-connect-runtime-0.0.1-SNAPSHOT.jar:na]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_221]
at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266) [na:1.8.0_221]
at java.util.concurrent.FutureTask.run(FutureTask.java) [na:1.8.0_221]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_221]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_221]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_221]
2019-12-22 22:49:00 INFO pool-4-thread-4 - Source task stop, config:{"properties":{"source-record-converter":"org.apache.rocketmq.connect.runtime.converter.JsonConverter","sourceRocketmq":"47.106.219.172:9876","task-class":"org.apache.rocketmq.replicator.RmqSourceTask","dataType":"0","storeTopic":"replicatorTopic","taskTopicList":"[{\"brokerName\":\"broker-b\",\"queueId\":6,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":4,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":2,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"},{\"brokerName\":\"broker-b\",\"queueId\":0,\"targetTopic\":\"fileTopic\",\"topic\":\"fileTopic\"}]","update-timestamp":"1577025620944"}}

```

Start with org.apache.rocketmq.replicator.RmqSourceTask.pollCommonMessage at RmqSourceTask.java:148 and the stop path shown in WorkerSourceTask.run at WorkerSourceTask.java:140. Reproduce by starting a Replicator with multiple tasks, then stopping the connector, and observe the SHUTDOWN_ALREADY error. Done means stopping the connector no longer reports this task poll error.

Written by the indexing model from the issue text.