最近學(xué)習(xí)flink的時(shí)候?qū)懥俗詈?jiǎn)單的一個(gè)demo,但是在測(cè)試的時(shí)候出現(xiàn)問題：
org.apache.flink.util.FlinkException: The TaskExecutor is shutting down.找了很久，也問了很多人基本沒人給出答案。

依賴

 <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <hadoop.version>3.1.1.3.1.0.0-78</hadoop.version>
        <flink.version>1.9.0</flink.version>
        <scala.binary.version>2.11</scala.binary.version>
        <scala.version>2.11.7</scala.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.7</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.45</version>
        </dependency>


        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.0.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-avro</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-jdbc_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

    </dependencies>

測(cè)試代碼

package com.meda.test

import java.util.Properties

import org.apache.flink.api.common.JobExecutionResult
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object FlinkKafka {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.enableCheckpointing(1000) 
    
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "GDGZ-TEL-DATACENTER01:6667,GDGZ-TEL-DATACENTER04:6667,GDGZ-TEL-DATACENTER05:6667")
    properties.setProperty("group.id", "test1")

    val myConsumer = new FlinkKafkaConsumer[String]("Tcollector", new SimpleStringSchema(), properties)
    myConsumer.setStartFromLatest() //從最新的記錄開始
    
    val stream: DataStream[String] = env
      .addSource(myConsumer)

    stream
      .print()
      .setParallelism(1)
    env.execute("streaming word count")
  }
}

遇到的問題

...
Initializing heap keyed state backend with stream factory.
[INFO ] 2019-11-25 17:09:47,701 method:org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.<init>(HeapKeyedStateBackend.java:137)
Initializing heap keyed state backend with stream factory.
[INFO ] 2019-11-25 17:09:47,701 method:org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.<init>(HeapKeyedStateBackend.java:137)
Initializing heap keyed state backend with stream factory.
.
.
.
{"data":{"nodetype":1,"session_id":7411},"nodeid":"logint","mid":9057551,"p_sub_typ_cd":0,"p_typ_cd":9,"time":1574715388,"uid":1000138974,"sys_time":"20191125205628747"}
{"data":{"nodetype":1,"session_id":7411},"nodeid":"ogout","mid":9057551,"p_sub_typ_cd":0,"p_typ_cd":9,"time":1574715388,"uid":1000140359,"sys_time":"20191125205628732"}
.
.
.
[INFO ] 2019-11-25 17:09:56,158 method:org.apache.flink.runtime.minicluster.MiniCluster.closeAsync(MiniCluster.java:424)
Shutting down Flink Mini Cluster
[INFO ] 2019-11-25 17:09:56,159 method:org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:322)
Stopping TaskExecutor akka://flink/user/taskmanager_0.
[INFO ] 2019-11-25 17:09:56,159 method:org.apache.flink.runtime.rest.RestServerEndpoint.closeAsync(RestServerEndpoint.java:290)
Shutting down rest endpoint.
[INFO ] 2019-11-25 17:09:56,159 method:org.apache.flink.runtime.executiongraph.Execution.transitionState(Execution.java:1419)
Flat Map -> Map (4/4) (1094fb6886b159077a6a2e168af1b4d1) switched from RUNNING to FAILED.
org.apache.flink.util.FlinkException: The TaskExecutor is shutting down.
    at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:332)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StartedState.terminate(AkkaRpcActor.java:504)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleControlMessage(AkkaRpcActor.java:170)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.onReceive(AkkaRpcActor.java:142)
    at akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:165)
    at akka.actor.Actor$class.aroundReceive(Actor.scala:502)
    at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:95)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
    at akka.actor.ActorCell.invoke(ActorCell.scala:495)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
    at akka.dispatch.Mailbox.run(Mailbox.scala:224)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
[INFO ] 2019-11-25 17:09:56,160 method:org.apache.flink.runtime.executiongraph.ExecutionGraph.transitionState(ExecutionGraph.java:1417)
Job Socket Window WordCount (e8df70bad57fbfeb80c17b8cf350501a) switched from state RUNNING to FAILING.
org.apache.flink.util.FlinkException: The TaskExecutor is shutting down.
    at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:332)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor$StartedState.terminate(AkkaRpcActor.java:504)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleControlMessage(AkkaRpcActor.java:170)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.onReceive(AkkaRpcActor.java:142)
    at akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:165)
    at akka.actor.Actor$class.aroundReceive(Actor.scala:502)
    at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:95)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
    at akka.actor.ActorCell.invoke(ActorCell.scala:495)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
    at akka.dispatch.Mailbox.run(Mailbox.scala:224)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
[INFO ] 2019-11-25 17:09:56,159 method:org.apache.flink.runtime.taskexecutor.JobLeaderService.stop(JobLeaderService.java:142)
Stop job leader service.
[INFO ] 2019-11-25 17:09:56,162 method:org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager.shutdown(TaskExecutorLocalStateStoresManager.java:213)
...

在執(zhí)行過程中勾缭，一開始流處理都可以執(zhí)行顾彰，但是幾秒后就停住了雀监，經(jīng)過debug和日志分析烈掠，Mini Cluster被停了且改。
但是相同的代碼女蜈、依賴發(fā)給以前的同事又可以執(zhí)行持舆，所以猜測(cè)是環(huán)境問題，所以嘗試各種環(huán)境的設(shè)置伪窖。

最后發(fā)現(xiàn)jdk版本有問題逸寓，我之前用得是jdk-8u181-linux-x64版本的,
由于一開始是正常的，完全沒往jdk想覆山。官網(wǎng)也沒具體說jdk1.8哪個(gè)版本竹伸。坑了好久簇宽，也問了很多人勋篓。

最終的解決辦法：

jdk 換成 jdk-8u231-windows-x64

jdk 換成 jdk-8u231-windows-x64

jdk 換成 jdk-8u231-windows-x64

image.png