Flink job消费kafka 失败，无法拿到offset值，请指教

有一个flink job在消费kafka topic消息，该topic存在于kafka两个集群cluster A 和Cluster B，Flink Job消费A集群的topic一切正常，但切换到消费B集群就启动失败。

Flink 集群采用Docker部署，Standalone模式。集群资源充足，Slot充足。报错日志如下：

java.lang.Exception: org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition Test-topic-27 could be determined at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowSourceExecutionException(SourceStreamTask.java:212) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.performDefaultAction(SourceStreamTask.java:132) at org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:298) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:403) at org.apache.flink.runtime.taskmanage有一个flink job在消费kafka topic消息，该topic存在于kafka两个集群cluster A 和Cluster B，Flink Job消费A集群的topic一切正常，但切换到消费B集群就启动失败。

Flink 集群采用Docker部署，Standalone模式。集群资源充足，Slot充足。报错日志如下：

java.lang.Exception: org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition Test-topic-27 could be determined at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowSourceExecutionException(SourceStreamTask.java:212) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.performDefaultAction(SourceStreamTask.java:132) at org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:298) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:403) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition Test-topic-27 could be determined

查询一圈发现基本都是说slot不够之类的原因，已经kafka broker负载等问题，这些问题已经排除。

请指教*来自志愿者整理的flinkr.Task.doRun(Task.java:705) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition Test-topic-27 could be determined

查询一圈发现基本都是说slot不够之类的原因，已经kafka broker负载等问题，这些问题已经排除。

请指教*来自志愿者整理的flink