技術標籤：flink大資料flinkdebug

在寫flink程式時報錯，後來網上查了好多文章都是瞎忽悠，最後自己仔細檢查程式碼發現，是自己的疏忽。

其實核心問題是你env 設定的時間語義和你程式碼中使用的不一致。

具體報錯資訊如下：

Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by FixedDelayRestartBackoffTimeStrategy(maxNumberRestartAttempts=2, backoffTimeMS=20000)
	at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:110)
	at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:76)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:192)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:186)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:180)
	at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:484)
	at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:380)
	at sun.reflect.GeneratedMethodAccessor19.invoke(Unknown Source)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:279)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:194)
	at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
	at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
	at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
	at akka.actor.Actor$class.aroundReceive(Actor.scala:517)
	at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
	at akka.actor.ActorCell.invoke(ActorCell.scala:561)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
	at akka.dispatch.Mailbox.run(Mailbox.scala:225)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
	... 4 more
Caused by: com.alibaba.druid.pool.DataSourceClosedException: dataSource already closed at Sun Feb 07 18:27:16 CST 2021
	at com.alibaba.druid.pool.DruidDataSource.getConnectionInternal(DruidDataSource.java:1108)
	at com.alibaba.druid.pool.DruidDataSource.getConnectionDirect(DruidDataSource.java:1012)
	at com.alibaba.druid.filter.FilterChainImpl.dataSource_connect(FilterChainImpl.java:4544)
	at com.alibaba.druid.filter.stat.StatFilter.dataSource_getConnection(StatFilter.java:662)
	at com.alibaba.druid.filter.FilterChainImpl.dataSource_connect(FilterChainImpl.java:4540)
	at com.alibaba.druid.pool.DruidDataSource.getConnection(DruidDataSource.java:990)
	at com.alibaba.druid.pool.DruidDataSource.getConnection(DruidDataSource.java:982)
	at com.beaver.experience.flink.java.utils.DBConnectionPoolOfInputUtil$.getConnection(DBConnectionPoolOfInputUtil.scala:25)
	at com.beaver.experience.flink.java.utils.DBConnectionPoolOfInputUtil.getConnection(DBConnectionPoolOfInputUtil.scala)
	at com.beaver.experience.flink.java.source.MySqlInputSource.run(MySqlInputSource.java:32)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:196)
 

提示這個：讓人一直把問題定位在重啟策略上。

最後解決辦法：

假如你程式碼中使用了EventTime，那麼你就在環境中設定他。

  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

如果你程式碼中用用 ProcessingTime，你就在env中設定

 env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

總之env中設定的時間語義要和程式碼中一致。