flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From John Stone <elliotst...@gmail.com>
Subject TaskManagers cannot contact JobManager in Kubernetes when JobManager HA is enabled
Date Wed, 31 Oct 2018 23:43:52 GMT
I have successfully managed to deploy a Flink cluster in Kubernetes without JobManager high
availability.  Everything works great.  The moment I enable high availability, TaskManagers
fail to contact the JobManager.  My configurations and logs are below.  Can someone point
me in the correct direction?  Many thanks!

Note the following:
- Cluster is running Flink 1.6.2 .
- Logs for the JobManager and TaskManagers indicate that they can successfully connect to
the Zookeeper instances.
- The Zookeeper instances reside as standard EC2 instances while the Flink cluster is fully
contained in Kubernetes.  Everything is within the same VPC.
- I have tried a TaskManager configuration which does not contain "high-availability: zookeeper"
(i.e. launched the JM under one configuration and the TMs under another such that this is
the only difference in the flink-conf.yaml file).  This did not help.

============================================================================

TaskManager Log Snippet

2018-10-31 23:14:09,035 INFO  akka.remote.transport.ProtocolStateActor                   
  - No response from remote for outbound association. Associate timed out after [20000 ms].
2018-10-31 23:14:09,035 WARN  akka.remote.ReliableDeliverySupervisor                     
  - Association with remote system [akka.tcp://flink@flink-jobmanager:43982] has failed, address
is now gated for [50] ms. Reason: [Association failed with [akka.tcp://flink@flink-jobmanager:43982]]
Caused by: [No response from remote for outbound association. Associate timed out after [20000
ms].]
2018-10-31 23:14:09,128 WARN  akka.remote.transport.netty.NettyTransport                 
  - Remote connection to [null] failed with org.apache.flink.shaded.akka.org.jboss.netty.channel.ConnectTimeoutException:
connection timed out: flink-jobmanager/100.70.34.197:43982
2018-10-31 23:14:19,135 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor         
  - Could not resolve ResourceManager address akka.tcp://flink@flink-jobmanager:43982/user/resourcemanager,
retrying in 10000 ms: Ask timed out on [ActorSelection[Anchor(akka.tcp://flink@flink-jobmanager:43982/),
Path(/user/resourcemanager)]] after [10000 ms]. Sender[null] sent message of type "akka.actor.Identify"..
2018-10-31 23:14:28,065 ERROR org.apache.flink.runtime.taskexecutor.TaskExecutor         
  - Fatal error occurred in TaskExecutor akka.tcp://flink@flink-taskmanager-75f746bdf7-fpw9h:38130/user/taskmanager_0.
org.apache.flink.runtime.taskexecutor.exceptions.RegistrationTimeoutException: Could not register
at the ResourceManager within the specified maximum registration duration 300000 ms. This
indicates a problem with this instance. Terminating now.

============================================================================

Configurations

JobManager Deployment

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: flink-jobmanager
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      containers:
      - name: jobmanager
        image: flink:1.6.2-hadoop28-scala_2.11
        command: ["/bin/sh", "-c", "cp /opt/flink/opt/flink-s3-fs-hadoop-1.6.2.jar /opt/flink/lib/flink-s3-fs-hadoop-1.6.2.jar
&& /docker-entrypoint.sh jobmanager"]
        ports:
        - containerPort: 6123
          name: rpc
        - containerPort: 6124
          name: blob
        - containerPort: 6125
          name: query
        - containerPort: 8081
          name: ui
        env:
        - name: JOB_MANAGER_RPC_ADDRESS
          value: flink-jobmanager
        - name: FLINK_CONF_DIR
          value: /etc/flink
        volumeMounts:
        - name: flink-config
          mountPath: /etc/flink
        - name: flink-hadoop-config
          mountPath: /etc/hadoop/conf
      volumes:
      - name: flink-config
        configMap:
          name: flink-config
      - name: flink-hadoop-config
        configMap:
          name: flink-hadoop-config

-----------------------------------------------------------------------------

JobManager Service

apiVersion: v1
kind: Service
metadata:
  name: flink-jobmanager
spec:
  ports:
  - name: rpc
    port: 6123
  - name: blob
    port: 6124
  - name: query
    port: 6125
  - name: ui
    port: 8081
  selector:
    app: flink
    component: jobmanager

-----------------------------------------------------------------------------

TaskManager Deployment

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: flink-taskmanager
spec:
  replicas: 2
  template:
    metadata:
      labels:
        app: flink
        component: taskmanager
    spec:
      containers:
      - name: taskmanager
        image: flink:1.6.2-hadoop28-scala_2.11
        command: ["/bin/sh", "-c", "cp /opt/flink/opt/flink-s3-fs-hadoop-1.6.2.jar /opt/flink/lib/flink-s3-fs-hadoop-1.6.2.jar
&& /docker-entrypoint.sh taskmanager"]
        ports:
        - containerPort: 6121
          name: data
        - containerPort: 6122
          name: rpc
        - containerPort: 6125
          name: query
        env:
        - name: JOB_MANAGER_RPC_ADDRESS
          value: flink-jobmanager
        - name: FLINK_CONF_DIR
          value: /etc/flink
        resources:
          limits:
            memory: "2Gi"
          requests:
            memory: "2Gi"
        volumeMounts:
        - name: flink-config
          mountPath: /etc/flink
        - name: flink-hadoop-config
          mountPath: /etc/hadoop/conf
      volumes:
      - name: flink-config
        configMap:
          name: flink-config
      - name: flink-hadoop-config
        configMap:
          name: flink-hadoop-config

-----------------------------------------------

Hadoop ConfigMap

apiVersion: v1
kind: ConfigMap
metadata:
  name: flink-hadoop-config
  labels:
    app: flink
data:
  core-site.xml: |
    <?xml version="1.0"?>
    <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
    <configuration>
      <property>
        <name>fs.s3.impl</name>
        <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
      </property>
      <property>
        <name>fs.s3a.buffer.dir</name>
        <value>/tmp</value>
      </property>
    </configuration>

------------------------------------------------

Flink ConfigMap

apiVersion: v1
kind: ConfigMap
metadata:
  name: flink-config
  labels:
    app: flink
data:
  flink-conf.yaml: |
    blob.server.port: 6124
    high-availability: zookeeper
    high-availability.zookeeper.quorum: zookeeper1:2181,zookeeper2:2181,zookeeper3:2181
    high-availability.storageDir: s3a://flink-test-bucket/flink-zk-storage
    high-availability.zookeeper.path.root: /flink-test
    high-availability.cluster-id: /flink-test
    jobmanager.heap.size: 1024m
    jobmanager.rpc.address: flink-jobmanager
    jobmanager.rpc.port: 6123
    parallelism.default: 1
    query.server.port: 6125
    rest.port: 8081
    state.backend.incremental: true
    state.backend: rocksdb
    state.checkpoints.dir: s3a://flink-test-bucket/flink-checkpoints
    taskmanager.heap.size: 1024m
    taskmanager.numberOfTaskSlots: 4
  log4j-console.properties: |
    log4j.rootLogger=INFO, console, file
    log4j.logger.akka=INFO
    log4j.logger.org.apache.kafka=INFO
    log4j.logger.org.apache.hadoop=INFO
    log4j.logger.org.apache.zookeeper=INFO
    log4j.logger.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, console
    log4j.appender.console=org.apache.log4j.ConsoleAppender
    log4j.appender.console.layout=org.apache.log4j.PatternLayout
    log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c
%x - %m%n

Mime
View raw message