roketmq can not consumer topic in cluster, it's ok in idea

# In standalone cluster
- environment:
> java version "1.8.0_271" + flink-1.14.6 + rocketmq 4.9.2
- code
```
// use
dataStream = env.addSource(createMqsource(cfg, dataSource));
// create RocketMQSource
public static RocketMQSource createMqsource(SourceCfg cfg, DataSourceCfg dataSource) {
// 构造mq地址
String[] hostNameSplit = dataSource.getHostname().split(SymbolConstant.SEPARATOR_SEMI_COLON);
String addr = null;
for (String s : hostNameSplit) {
String joinStr = s + SymbolConstant.SEPARATOR_COLON + dataSource.getPort();
addr = StringUtils.isEmpty(addr) ? joinStr : addr + SymbolConstant.SEPARATOR_SEMI_COLON + joinStr;
}
DeserializationSchema, TransData> deserializationSchema = MqDeserializationSchemaFactory.create(cfg);
// 判断启动模式
OffsetResetStrategy startUpMode = cfg.getStartupMode() == null || cfg.getStartupMode().equals(StartupModeEnum.LATEST) ?
OffsetResetStrategy.LATEST : OffsetResetStrategy.EARLIEST;

RocketMQSourceBuilder builder = new RocketMQSourceBuilder()
.setNameServerAddress(addr)
.setConsumerGroup(cfg.getConsumerGroup())
.setTopic(cfg.getTopic())
.setTag(cfg.getTag())
.setStartFromGroupOffsets(startUpMode)
.setDeserializer(new RocketMQValueOnlyDeserializationSchemaWrapper<>(deserializationSchema));
log.info("addr:{}, group:{}, tag:{}", addr, cfg.getConsumerGroup(), cfg.getTag());
return builder.build();
}
```
- exception infos:

2024-01-05 13:48:01,579 INFO org.apache.rocketmq.flink.source.enumerator.RocketMQSourceEnumerator [] - Starting the RocketMQSourceEnumerator for consumer group X without periodic partition discovery.

2024-01-05 13:48:01,589 INFO org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Source Source: T_AC_RE_WZ_JJJZ(MQ)- **_registering reader for parallel task 0 @ 127.0.0.1_**
2024-01-05 13:48:01,886 INFO RocketmqRemoting [] - closeChannel: close the connection to remote address[10.200.38.118:9876] result: true
2024-01-05 13:48:01,887 INFO RocketmqRemoting [] - closeChannel: close the connection to remote address[10.200.38.118:9876] result: true
2024-01-05 13:48:01,889 INFO RocketmqRemoting [] - closeChannel: close the connection to remote address[10.200.38.118:9876] result: true
2024-01-05 13:48:01,889 INFO RocketmqRemoting [] - closeChannel: close the connection to remote address[10.200.38.118:9876] result: true
2024-01-05 13:48:01,890 INFO RocketmqRemoting [] - closeChannel: close the connection to remote address[10.200.38.118:9876] result: true
2024-01-05 13:48:01,887 ERROR org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext [] - Exception while handling result from async call in SourceCoordinator-Source: S(MQ)-. Triggering job failover.
org.apache.flink.util.FlinkRuntimeException: _**Failed to handle partition splits change due to**_
at org.apache.rocketmq.flink.source.enumerator.RocketMQSourceEnumerator.handlePartitionSplitChanges(RocketMQSourceEnumerator.java:279) ~[flink-rocketmq-1.0-SNAPSHOT.jar:?]
at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$null$1(ExecutorNotifier.java:83) ~[flink-runtime-1.14.6.jar:1.14.6]
at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40) [flink-core-1.14.6.jar:1.14.6]
at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) [?:1.8.0_271]
at java.util.concurrent.FutureTask.run(Unknown Source) [?:1.8.0_271]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(Unknown Source) [?:1.8.0_271]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source) [?:1.8.0_271]
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) [?:1.8.0_271]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) [?:1.8.0_271]
at java.lang.Thread.run(Unknown Source) [?:1.8.0_271]
Caused by: org.apache.rocketmq.client.exception.MQClientException: _**Can not find Message Queue for this topic**_, FPC_SYNC_FI_AC_REPTILE_1
See http://rocketmq.apache.org/docs/faq/ for further details.
at org.apache.rocketmq.client.impl.MQAdminImpl.fetchSubscribeMessageQueues(MQAdminImpl.java:177) ~[rocketmq-client-4.9.2.jar:4.9.2]
at org.apache.rocketmq.client.impl.consumer.DefaultMQPullConsumerImpl.fetchSubscribeMessageQueues(DefaultMQPullConsumerImpl.java:147) ~[rocketmq-client-4.9.2.jar:4.9.2]
at org.apache.rocketmq.client.consumer.DefaultMQPullConsumer.fetchSubscribeMessageQueues(DefaultMQPullConsumer.java:290) ~[rocketmq-client-4.9.2.jar:4.9.2]
at org.apache.rocketmq.flink.source.enumerator.RocketMQSourceEnumerator.discoverAndInitializePartitionSplit(RocketMQSourceEnumerator.java:248) ~[flink-rocketmq-1.0-SNAPSHOT.jar:?]
at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$notifyReadyAsync$2(ExecutorNotifier.java:80) ~[flink-runtime-1.14.6.jar:1.14.6]
... 7 more
Caused by: org.apache.rocketmq.remoting.exception.RemotingSendRequestException: _**send request to <10.200.38.118:9876> failed**_
at org.apache.rocketmq.remoting.netty.NettyRemotingAbstract.invokeSyncImpl(NettyRemotingAbstract.java:440) ~[rocketmq-remoting-4.9.2.jar:4.9.2]
at org.apache.rocketmq.remoting.netty.NettyRemotingClient.invokeSync(NettyRemotingClient.java:377) ~[rocketmq-remoting-4.9.2.jar:4.9.2]
at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1367) ~[rocketmq-client-4.9.2.jar:4.9.2]
at org.apache.rocketmq.client.impl.MQClientAPIImpl.getTopicRouteInfoFromNameServer(MQClientAPIImpl.java:1357) ~[rocketmq-client-4.9.2.jar:4.9.2]
at org.apache.rocketmq.client.impl.MQAdminImpl.fetchSubscribeMessageQueues(MQAdminImpl.java:166) ~[rocketmq-client-4.9.2.jar:4.9.2]
at org.apache.rocketmq.client.impl.consumer.DefaultMQPullConsumerImpl.fetchSubscribeMessageQueues(DefaultMQPullConsumerImpl.java:147) ~[rocketmq-client-4.9.2.jar:4.9.2]
at org.apache.rocketmq.client.consumer.DefaultMQPullConsumer.fetchSubscribeMessageQueues(DefaultMQPullConsumer.java:290) ~[rocketmq-client-4.9.2.jar:4.9.2]
at org.apache.rocketmq.flink.source.enumerator.RocketMQSourceEnumerator.discoverAndInitializePartitionSplit(RocketMQSourceEnumerator.java:248) ~[flink-rocketmq-1.0-SNAPSHOT.jar:?]
at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$notifyReadyAsync$2(ExecutorNotifier.java:80) ~[flink-runtime-1.14.6.jar:1.14.6]
... 7 more

Start with RocketMQSourceEnumerator.discoverAndInitializePartitionSplit at line 248 and handlePartitionSplitChanges at line 279, then trace the RocketMQ client call fetching topic queues. Reproduce the failure with the stated Java, Flink, RocketMQ, cluster setup and inspect the NameServer connection errors. Done means the source can discover and consume the topic in the cluster without failing over.

Written by the indexing model from the issue text.