Flink消费Kafka报错：SplitFetcher thread 0 received unexpected exceptio ...

报错信息如下：

1. org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
2.         at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:139) ~[flink-runtime-1.17.1.jar:1.17.1]
3.         at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:83) ~[flink-runtime-1.17.1.jar:1.17.1]
4.         at org.apache.flink.runtime.scheduler.DefaultScheduler.recordTaskFailure(DefaultScheduler.java:258) ~[flink-runtime-1.17.1.jar:1.17.1]
5.         at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:249) ~[flink-runtime-1.17.1.jar:1.17.1]
6.         at org.apache.flink.runtime.scheduler.DefaultScheduler.onTaskFailed(DefaultScheduler.java:242) ~[flink-runtime-1.17.1.jar:1.17.1]
7.         at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:748) ~[flink-runtime-1.17.1.jar:1.17.1]
8.         at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:725) ~[flink-runtime-1.17.1.jar:1.17.1]
9.         at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:80) ~[flink-runtime-1.17.1.jar:1.17.1]
10.         at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:479) ~[flink-runtime-1.17.1.jar:1.17.1]
11.         at jdk.internal.reflect.GeneratedMethodAccessor23.invoke(Unknown Source) ~[na:na]
12.         at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
13.         at java.base/java.lang.reflect.Method.invoke(Method.java:566) ~[na:na]
14.         at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRpcInvocation$1(AkkaRpcActor.java:309) ~[na:na]
15.         at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83) ~[na:na]
16.         at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:307) ~[na:na]
17.         at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:222) ~[na:na]
18.         at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:84) ~[na:na]
19.         at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:168) ~[na:na]
20.         at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24) ~[na:na]
21.         at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20) ~[na:na]
22.         at scala.PartialFunction.applyOrElse(PartialFunction.scala:127) ~[na:na]
23.         at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126) ~[na:na]
24.         at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20) ~[na:na]
25.         at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175) ~[na:na]
26.         at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176) ~[na:na]
27.         at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176) ~[na:na]
28.         at akka.actor.Actor.aroundReceive(Actor.scala:537) ~[na:na]
29.         at akka.actor.Actor.aroundReceive$(Actor.scala:535) ~[na:na]
30.         at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220) ~[na:na]
31.         at akka.actor.ActorCell.receiveMessage$$$capture(ActorCell.scala:579) ~[na:na]
32.         at akka.actor.ActorCell.receiveMessage(ActorCell.scala) ~[na:na]
33.         at akka.actor.ActorCell.invoke(ActorCell.scala:547) ~[na:na]
34.         at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270) ~[na:na]
35.         at akka.dispatch.Mailbox.run(Mailbox.scala:231) ~[na:na]
36.         at akka.dispatch.Mailbox.exec(Mailbox.scala:243) ~[na:na]
37.         at java.base/java.util.concurrent.ForkJoinTask.doExec$$$capture(ForkJoinTask.java:290) ~[na:na]
38.         at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java) ~[na:na]
39.         at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020) ~[na:na]
40.         at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656) ~[na:na]
41.         at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594) ~[na:na]
42.         at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183) ~[na:na]
43. Caused by: java.lang.RuntimeException: One or more fetchers have encountered exception
44.         at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:261) ~[flink-connector-base-1.17.1.jar:1.17.1]
45.         at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:169) ~[flink-connector-base-1.17.1.jar:1.17.1]
46.         at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:131) ~[flink-connector-base-1.17.1.jar:1.17.1]
47.         at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:417) ~[flink-streaming-java-1.17.1.jar:1.17.1]
48.         at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68) ~[flink-streaming-java-1.17.1.jar:1.17.1]
49.         at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-streaming-java-1.17.1.jar:1.17.1]
50.         at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:550) ~[flink-streaming-java-1.17.1.jar:1.17.1]
51.         at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-streaming-java-1.17.1.jar:1.17.1]
52.         at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:839) ~[flink-streaming-java-1.17.1.jar:1.17.1]
53.         at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:788) ~[flink-streaming-java-1.17.1.jar:1.17.1]
54.         at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952) ~[flink-runtime-1.17.1.jar:1.17.1]
55.         at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931) ~[flink-runtime-1.17.1.jar:1.17.1]
56.         at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745) ~[flink-runtime-1.17.1.jar:1.17.1]
57.         at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562) ~[flink-runtime-1.17.1.jar:1.17.1]
58.         at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]
59. Caused by: java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records
60.         at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165) ~[flink-connector-base-1.17.1.jar:1.17.1]
61.         at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:114) ~[flink-connector-base-1.17.1.jar:1.17.1]
62.         at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
63.         at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264) ~[na:na]
64.         at java.base/java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:na]
65.         at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
66.         at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
67.         ... 1 common frames omitted
68. Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition device_meter_data-0 could be determined

复制代码

上面的信息是在flink中展示报错信息。
查看Kafka服务端，又发现如下错误信息

1. Error processing create topic request CreatableTopic(name='__consumer_offsets', numPartitions=50, replicationFactor=3, assignments=[], configs=[CreateableTopicConfig(name='compression.type', value='producer'), CreateableTopicConfig(name='cleanup.policy', value='compact'), CreateableTopicConfig(name='segment.bytes', value='104857600')]) (kafka.server.ZkAdminManager)
2. org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 3 larger than available brokers: 1.

复制代码

解决方案：

面对 InvalidReplicationFactorException 非常，其核心问题是实验设定的复制因子（replication factor）超出了 Kafka 集群中可用的 broker 数量。针对您提供的错误信息，解决方案在于调整复制因子以适应现有的 broker 数量。
解决方案步骤

核查 Kafka 集群的 Broker 数量：
确保您清楚集群中生动的 broker 数量。若集群仅包罗一个 broker，复制因子必须设为 1。
修改复制因子设置：
根据集群的 broker 数量，调整主题的复制因子。若 broker 数量为 1，则应将复制因子设为 1。
重新创建主题：
利用修改后的复制因子设置，重新实验创建主题。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。