Thanks! Till. I'll give a try on your suggestions and update the thread.
On Wed, May 23, 2018 at 4:43 AM, Till Rohrmann <trohrm...@apache.org> wrote: > Hi Amit, > > it looks as if the current cancellation cause is not the same as the > initially reported cancellation cause. In the current case, it looks as if > the deployment of your tasks takes so long that that maximum > `taskmanager.network.request-backoff.max` value has been reached. When this > happens a task gives up asking for the input result partition and fails with > the `PartitionNotFoundException`. > > More concretely, the `CHAIN Reduce (GroupReduce at > first(SortedGrouping.java:210)) -> Map (Key Extractor) (2/14)` cannot > retrieve the result partition of the `CHAIN DataSource (at > createInput(ExecutionEnvironment.java:548) > (org.apache.flink.api.java.io.TextInputFormat)) -> Map (Data Source > org.apache.flink.api.java.io.TextInputFormat > [s3a://limeroad-logs/mysqlbin_lradmin_s2d_order_data2/redshift_logs/2018/5/20/14/, > s3a://limeroad-logs/mysqlbin_lradmin_s2d_order_data2/redshift_logs/2018/5/20/15/0/]) > -> Map (Key Extractor) -> Combine (GroupReduce at > first(SortedGrouping.java:210)) (8/14)` task. This tasks is in the state > deploying when the exception occurs. It seems to me that this task takes > quite some time to be deployed. > > One reason why the deployment could take some time is that an UDF (for > example the closure) of one of the operators is quite large. If this is the > case, then Flink offloads the corresponding data onto the BlobServer from > where they are retrieved by the TaskManagers. Since you are running in > non-HA mode, the BlobServer won't store the blobs on HDFS from where they > could be retrieved. Instead all the TaskManagers ask the single BlobServer > for the required TDD blobs. Depending on the size of the TDDs, the > BlobServer might become the bottleneck. > > What you can try to do is the following > 1) Try to reduce the closure object of the UDFs in the above-mentioned task. > 2) Increase `taskmanager.network.request-backoff.max` to give the system > more time to download the blobs > 3) Run the cluster in HA mode which will store the blobs also under > `high-availability.storageDir` (usually HDFS or S3). Before downloading the > blobs from the BlobServer, Flink will first try to download them from the > `high-availability-storageDir` > > Let me know if this solves your problem. > > Cheers, > Till > > On Tue, May 22, 2018 at 1:29 PM, Amit Jain <aj201...@gmail.com> wrote: >> >> Hi Nico, >> >> Please find the attachment for more logs. >> >> -- >> Thanks, >> Amit >> >> On Tue, May 22, 2018 at 4:09 PM, Nico Kruber <n...@data-artisans.com> >> wrote: >> > Hi Amit, >> > thanks for providing the logs, I'll look into it. We currently have a >> > suspicion of this being caused by >> > https://issues.apache.org/jira/browse/FLINK-9406 which we found by >> > looking over the surrounding code. The RC4 has been cancelled since we >> > see this as a release blocker. >> > >> > To rule out further errors, can you also provide logs for the task >> > manager producing partitions d6946b39439f10e8189322becf1b8887, >> > 9aa35dd53561f9220b7b1bad84309d5f and 60909dec9134eb980e18064358dc2c81? >> > The task manager log you provided covers the task manager asking for >> > this partition only for which the job manager produces the >> > PartitionProducerDisposedException that you see. >> > I'm looking for the logs of task managers with the following execution >> > IDs in their logs: >> > - 2826f9d430e05e9defaa46f60292fa79 >> > - 7ef992a067881a07409819e3aea32004 >> > - ec923ce6d891d89cf6fecb5e3f5b7cc5 >> > >> > Regarding the operators being stuck: I'll have a further look into the >> > logs and state transition and will come back to you. >> > >> > >> > Nico >> > >> > >> > On 21/05/18 09:27, Amit Jain wrote: >> >> Hi All, >> >> >> >> I totally missed this thread. I've encountered same issue in Flink >> >> 1.5.0 RC4. Please look over the attached logs of JM and impacted TM. >> >> >> >> Job ID 390a96eaae733f8e2f12fc6c49b26b8b >> >> >> >> -- >> >> Thanks, >> >> Amit >> >> >> >> On Thu, May 3, 2018 at 8:31 PM, Nico Kruber <n...@data-artisans.com> >> >> wrote: >> >>> Also, please have a look at the other TaskManagers' logs, in >> >>> particular >> >>> the one that is running the operator that was mentioned in the >> >>> exception. You should look out for the ID >> >>> 98f5976716234236dc69fb0e82a0cc34. >> >>> >> >>> >> >>> Nico >> >>> >> >>> >> >>> PS: Flink logs files should compress quite nicely if they grow too big >> >>> :) >> >>> >> >>> On 03/05/18 14:07, Stephan Ewen wrote: >> >>>> Google Drive would be great. >> >>>> >> >>>> Thanks! >> >>>> >> >>>> On Thu, May 3, 2018 at 1:33 PM, Amit Jain <aj201...@gmail.com >> >>>> <mailto:aj201...@gmail.com>> wrote: >> >>>> >> >>>> Hi Stephan, >> >>>> >> >>>> Size of JM log file is 122 MB. Could you provide me other media >> >>>> to >> >>>> post the same? We can use Google Drive if that's fine with you. >> >>>> >> >>>> -- >> >>>> Thanks, >> >>>> Amit >> >>>> >> >>>> On Thu, May 3, 2018 at 12:58 PM, Stephan Ewen <se...@apache.org >> >>>> <mailto:se...@apache.org>> wrote: >> >>>> > Hi Amit! >> >>>> > >> >>>> > Thanks for sharing this, this looks like a regression with the >> >>>> network stack >> >>>> > changes. >> >>>> > >> >>>> > The log you shared from the TaskManager gives some hint, but >> >>>> that >> >>>> exception >> >>>> > alone should not be a problem. That exception can occur under a >> >>>> race between >> >>>> > deployment of some tasks while the whole job is entering a >> >>>> recovery phase >> >>>> > (maybe we should not print it so prominently to not confuse >> >>>> users). There >> >>>> > must be something else happening on the JobManager. Can you >> >>>> share >> >>>> the JM >> >>>> > logs as well? >> >>>> > >> >>>> > Thanks a lot, >> >>>> > Stephan >> >>>> > >> >>>> > >> >>>> > On Wed, May 2, 2018 at 12:21 PM, Amit Jain <aj201...@gmail.com >> >>>> <mailto:aj201...@gmail.com>> wrote: >> >>>> >> >> >>>> >> Thanks! Fabian >> >>>> >> >> >>>> >> I will try using the current release-1.5 branch and update >> >>>> this >> >>>> thread. >> >>>> >> >> >>>> >> -- >> >>>> >> Thanks, >> >>>> >> Amit >> >>>> >> >> >>>> >> On Wed, May 2, 2018 at 3:42 PM, Fabian Hueske >> >>>> <fhue...@gmail.com >> >>>> <mailto:fhue...@gmail.com>> wrote: >> >>>> >> > Hi Amit, >> >>>> >> > >> >>>> >> > We recently fixed a bug in the network stack that affected >> >>>> batch jobs >> >>>> >> > (FLINK-9144). >> >>>> >> > The fix was added after your commit. >> >>>> >> > >> >>>> >> > Do you have a chance to build the current release-1.5 branch >> >>>> and check >> >>>> >> > if >> >>>> >> > the fix also resolves your problem? >> >>>> >> > >> >>>> >> > Otherwise it would be great if you could open a blocker >> >>>> issue >> >>>> for the >> >>>> >> > 1.5 >> >>>> >> > release to ensure that this is fixed. >> >>>> >> > >> >>>> >> > Thanks, >> >>>> >> > Fabian >> >>>> >> > >> >>>> >> > 2018-04-29 18:30 GMT+02:00 Amit Jain <aj201...@gmail.com >> >>>> <mailto:aj201...@gmail.com>>: >> >>>> >> >> >> >>>> >> >> Cluster is running on commit 2af481a >> >>>> >> >> >> >>>> >> >> On Sun, Apr 29, 2018 at 9:59 PM, Amit Jain >> >>>> <aj201...@gmail.com >> >>>> <mailto:aj201...@gmail.com>> wrote: >> >>>> >> >> > Hi, >> >>>> >> >> > >> >>>> >> >> > We are running numbers of batch jobs in Flink 1.5 cluster >> >>>> and few of >> >>>> >> >> > those >> >>>> >> >> > are getting stuck at random. These jobs having the >> >>>> following >> >>>> failure >> >>>> >> >> > after >> >>>> >> >> > which operator status changes to CANCELED and stuck to >> >>>> same. >> >>>> >> >> > >> >>>> >> >> > Please find complete TM's log at >> >>>> >> >> > >> >>>> >> >>>> https://gist.github.com/imamitjain/066d0e99990ee24f2da1ddc83eba2012 >> >>>> >> >>>> <https://gist.github.com/imamitjain/066d0e99990ee24f2da1ddc83eba2012> >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >> > 2018-04-29 14:57:24,437 INFO >> >>>> >> >> > org.apache.flink.runtime.taskmanager.Task >> >>>> >> >> > - Producer 98f5976716234236dc69fb0e82a0cc34 of partition >> >>>> >> >> > 5b8dd5ac2df14266080a8e049defbe38 disposed. Cancelling >> >>>> execution. >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >>>> org.apache.flink.runtime.jobmanager.PartitionProducerDisposedException: >> >>>> >> >> > Execution 98f5976716234236dc69fb0e82a0cc34 producing >> >>>> partition >> >>>> >> >> > 5b8dd5ac2df14266080a8e049defbe38 has already been >> >>>> disposed. >> >>>> >> >> > at >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >>>> org.apache.flink.runtime.jobmaster.JobMaster.requestPartitionState(JobMaster.java:610) >> >>>> >> >> > at sun.reflect.GeneratedMethodAccessor107.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:210) >> >>>> >> >> > at >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >>>> org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:154) >> >>>> >> >> > at >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >>>> org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleMessage(FencedAkkaRpcActor.java:69) >> >>>> >> >> > at >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >>>> org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$onReceive$1(AkkaRpcActor.java:132) >> >>>> >> >> > at >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >>>> akka.actor.ActorCell$$anonfun$become$1.applyOrElse(ActorCell.scala:544) >> >>>> >> >> > 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) >> >>>> >> >> > >> >>>> >> >> > >> >>>> >> >> > Thanks >> >>>> >> >> > Amit >> >>>> >> > >> >>>> >> > >> >>>> > >> >>>> > >> >>>> >> >>>> >> >>> >> >>> -- >> >>> Nico Kruber | Software Engineer >> >>> data Artisans >> >>> >> >>> Follow us @dataArtisans >> >>> -- >> >>> Join Flink Forward - The Apache Flink Conference >> >>> Stream Processing | Event Driven | Real Time >> >>> -- >> >>> Data Artisans GmbH | Stresemannstr. 121A,10963 Berlin, Germany >> >>> data Artisans, Inc. | 1161 Mission Street, San Francisco, CA-94103, >> >>> USA >> >>> -- >> >>> Data Artisans GmbH >> >>> Registered at Amtsgericht Charlottenburg: HRB 158244 B >> >>> Managing Directors: Dr. Kostas Tzoumas, Dr. Stephan Ewen >> >>> >> > >> > -- >> > Nico Kruber | Software Engineer >> > data Artisans >> > >> > Follow us @dataArtisans >> > -- >> > Join Flink Forward - The Apache Flink Conference >> > Stream Processing | Event Driven | Real Time >> > -- >> > Data Artisans GmbH | Stresemannstr. 121A,10963 Berlin, Germany >> > data Artisans, Inc. | 1161 Mission Street, San Francisco, CA-94103, USA >> > -- >> > Data Artisans GmbH >> > Registered at Amtsgericht Charlottenburg: HRB 158244 B >> > Managing Directors: Dr. Kostas Tzoumas, Dr. Stephan Ewen >> > > >
