Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
MIME-Version: 1.0
From: Geoffrey Mon <geofbot@gmail.com>
Date: Thu, 09 Feb 2017 23:34:51 +0000
Message-ID: <CAES76rFoFTrxnjZkFwwXnk0L1VnDO4UTTtT2_=kiT=+0wkXGqQ@mail.gmail.com>
Subject: "Job submission to the JobManager timed out" on EMR YARN cluster with
 multiple jobs
To: "user@flink.apache.org" <user@flink.apache.org>
Content-Type: multipart/alternative; boundary=94eb2c0712721ccb490548216e4a
archived-at: Thu, 09 Feb 2017 23:35:12 -0000

--94eb2c0712721ccb490548216e4a
Content-Type: text/plain; charset=UTF-8

Hello all,

I'm running a Flink plan made up of multiple jobs. The source for my job
can be found here if it would help in any way:
https://github.com/quinngroup/flink-r1dl/blob/master/src/main/java/com/github/quinngroup/R1DL.java
Each of the jobs (except for the first job) depends on files generated by
the previous job; I'm running it on an AWS EMR cluster using YARN.

When I submit the plan file, the first job runs as planned. After it
completes, the second job is submitted by the YARN client:

<snip>
02/09/2017 16:39:43 DataSink (CsvSink)(4/5) switched to FINISHED
02/09/2017 16:39:43 Job execution switched to status FINISHED.
2017-02-09 16:40:26,470 INFO  org.apache.flink.yarn.YarnClusterClient
                - Waiting until all TaskManagers have connected
Waiting until all TaskManagers have connected
2017-02-09 16:40:26,476 INFO  org.apache.flink.yarn.YarnClusterClient
                - TaskManager status (5/5)
TaskManager status (5/5)
2017-02-09 16:40:26,476 INFO  org.apache.flink.yarn.YarnClusterClient
                - All TaskManagers are connected
All TaskManagers are connected
2017-02-09 16:40:26,480 INFO  org.apache.flink.yarn.YarnClusterClient
                - Submitting job with JobID:
b226f5f18a78bc386bd1b1b6d30515ea. Waiting for job completion.
Submitting job with JobID: b226f5f18a78bc386bd1b1b6d30515ea. Waiting for
job completion.
Connected to JobManager at Actor[akka.tcp://flink@
<snip>.ec2.internal:35598/user/jobmanager#68430682]

If the input file is small and the first job runs quickly (~1 minute works
for me), then the second job runs fine. However, if the input file for my
first job is large and the first job takes more than a minute or so to
complete, Flink will not acknowledge receiving the next job; the web Flink
console does not show any new jobs and Flink logs do not mention receiving
any new jobs after the first job has completed. The YARN client's job
submission times out after Flink does not respond:

Caused by:
org.apache.flink.runtime.client.JobClientActorSubmissionTimeoutException:
Job submission to the JobManager timed out. You may increase
'akka.client.timeout' in case the JobManager needs more time to configure
and confirm the job submission.
at
org.apache.flink.runtime.client.JobSubmissionClientActor.handleCustomMessage(JobSubmissionClientActor.java:119)
at
org.apache.flink.runtime.client.JobClientActor.handleMessage(JobClientActor.java:239)
at
org.apache.flink.runtime.akka.FlinkUntypedActor.handleLeaderSessionID(FlinkUntypedActor.java:88)
at
org.apache.flink.runtime.akka.FlinkUntypedActor.onReceive(FlinkUntypedActor.java:68)
at
akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:167)

I have tried increasing akka.client.timeout to large values such as 1200s
(20 minutes), but even then Flink does not acknowledge or execute any other
jobs and there is the same timeout error. Does anyone know how I can get
Flink to execute all of the jobs properly?

Cheers,
Geoffrey Mon

--94eb2c0712721ccb490548216e4a
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr">Hello all,<div><br></div><div>I&#39;m running a Flink plan=
 made up of multiple jobs. The source for my job can be found here if it wo=
uld help in any way:=C2=A0<a href=3D"https://github.com/quinngroup/flink-r1=
dl/blob/master/src/main/java/com/github/quinngroup/R1DL.java">https://githu=
b.com/quinngroup/flink-r1dl/blob/master/src/main/java/com/github/quinngroup=
/R1DL.java</a></div><div>Each of the jobs (except for the first job) depend=
s on files generated by the previous job; I&#39;m running it on an AWS EMR =
cluster using YARN.</div><div><br></div><div>When I submit the plan file, t=
he first job runs as planned. After it completes, the second job is submitt=
ed by the YARN client:</div><div><br></div><div><div>&lt;snip&gt;</div><div=
>02/09/2017 16:39:43<span class=3D"Apple-tab-span" style=3D"white-space:pre=
">	</span>DataSink (CsvSink)(4/5) switched to FINISHED=C2=A0</div><div>02/0=
9/2017 16:39:43<span class=3D"Apple-tab-span" style=3D"white-space:pre">	</=
span>Job execution switched to status FINISHED.</div><div>2017-02-09 16:40:=
26,470 INFO =C2=A0org.apache.flink.yarn.YarnClusterClient =C2=A0 =C2=A0 =C2=
=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 - Waiting until=
 all TaskManagers have connected</div><div>Waiting until all TaskManagers h=
ave connected</div><div>2017-02-09 16:40:26,476 INFO =C2=A0org.apache.flink=
.yarn.YarnClusterClient =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =
=C2=A0 =C2=A0 =C2=A0 =C2=A0 - TaskManager status (5/5)</div><div>TaskManage=
r status (5/5)</div><div>2017-02-09 16:40:26,476 INFO =C2=A0org.apache.flin=
k.yarn.YarnClusterClient =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =
=C2=A0 =C2=A0 =C2=A0 =C2=A0 - All TaskManagers are connected</div><div>All =
TaskManagers are connected</div><div>2017-02-09 16:40:26,480 INFO =C2=A0org=
.apache.flink.yarn.YarnClusterClient =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=
=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 - Submitting job with JobID: b226f5f=
18a78bc386bd1b1b6d30515ea. Waiting for job completion.</div><div>Submitting=
 job with JobID: b226f5f18a78bc386bd1b1b6d30515ea. Waiting for job completi=
on.</div><div>Connected to JobManager at Actor[akka.tcp://flink@&lt;snip&gt=
;.ec2.internal:35598/user/jobmanager#68430682]</div></div><div><br></div><d=
iv>If the input file is small and the first job runs quickly (~1 minute wor=
ks for me), then the second job runs fine. However, if the input file for m=
y first job is large and the first job takes more than a minute or so to co=
mplete, Flink will not acknowledge receiving the next job; the web Flink co=
nsole does not show any new jobs and Flink logs do not mention receiving an=
y new jobs after the first job has completed. The YARN client&#39;s job sub=
mission times out after Flink does not respond:</div><div><br></div><div><d=
iv>Caused by: org.apache.flink.runtime.client.JobClientActorSubmissionTimeo=
utException: Job submission to the JobManager timed out. You may increase &=
#39;akka.client.timeout&#39; in case the JobManager needs more time to conf=
igure and confirm the job submission.</div><div><span class=3D"Apple-tab-sp=
an" style=3D"white-space:pre">	</span>at org.apache.flink.runtime.client.Jo=
bSubmissionClientActor.handleCustomMessage(JobSubmissionClientActor.java:11=
9)</div><div><span class=3D"Apple-tab-span" style=3D"white-space:pre">	</sp=
an>at org.apache.flink.runtime.client.JobClientActor.handleMessage(JobClien=
tActor.java:239)</div><div><span class=3D"Apple-tab-span" style=3D"white-sp=
ace:pre">	</span>at org.apache.flink.runtime.akka.FlinkUntypedActor.handleL=
eaderSessionID(FlinkUntypedActor.java:88)</div><div><span class=3D"Apple-ta=
b-span" style=3D"white-space:pre">	</span>at org.apache.flink.runtime.akka.=
FlinkUntypedActor.onReceive(FlinkUntypedActor.java:68)</div><div><span clas=
s=3D"Apple-tab-span" style=3D"white-space:pre">	</span>at akka.actor.Untype=
dActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:167)</div></div><d=
iv><br></div><div>I have tried increasing akka.client.timeout to large valu=
es such as 1200s (20 minutes), but even then Flink does not acknowledge or =
execute any other jobs and there is the same timeout error. Does anyone kno=
w how I can get Flink to execute all of the jobs properly?</div><div><br></=
div><div>Cheers,</div><div>Geoffrey Mon</div></div>

--94eb2c0712721ccb490548216e4a--