From user-return-26493-archive-asf-public=cust-asf.ponee.io@flink.apache.org Wed Mar 13 12:19:20 2019 Return-Path: X-Original-To: archive-asf-public@cust-asf.ponee.io Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by mx-eu-01.ponee.io (Postfix) with SMTP id A2A2118064A for ; Wed, 13 Mar 2019 13:19:19 +0100 (CET) Received: (qmail 56148 invoked by uid 500); 13 Mar 2019 12:19:18 -0000 Mailing-List: contact user-help@flink.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Delivered-To: mailing list user@flink.apache.org Received: (qmail 56137 invoked by uid 99); 13 Mar 2019 12:19:18 -0000 Received: from pnap-us-west-generic-nat.apache.org (HELO spamd1-us-west.apache.org) (209.188.14.142) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 13 Mar 2019 12:19:18 +0000 Received: from localhost (localhost [127.0.0.1]) by spamd1-us-west.apache.org (ASF Mail Server at spamd1-us-west.apache.org) with ESMTP id E4731C0F6F for ; Wed, 13 Mar 2019 12:19:17 +0000 (UTC) X-Virus-Scanned: Debian amavisd-new at spamd1-us-west.apache.org X-Spam-Flag: NO X-Spam-Score: -0.301 X-Spam-Level: X-Spam-Status: No, score=-0.301 tagged_above=-999 required=6.31 tests=[HTML_MESSAGE=2, RCVD_IN_DNSWL_MED=-2.3, SPF_PASS=-0.001] autolearn=disabled Received: from mx1-lw-eu.apache.org ([10.40.0.8]) by localhost (spamd1-us-west.apache.org [10.40.0.7]) (amavisd-new, port 10024) with ESMTP id x12bo54IvHUB for ; Wed, 13 Mar 2019 12:19:15 +0000 (UTC) Received: from mailservice.tudelft.nl (mailservice.tudelft.nl [130.161.131.5]) by mx1-lw-eu.apache.org (ASF Mail Server at mx1-lw-eu.apache.org) with ESMTPS id 27CF55F11F for ; Wed, 13 Mar 2019 12:19:15 +0000 (UTC) Received: from localhost (localhost [127.0.0.1]) by amavis (Postfix) with ESMTP id 12E6C640575 for ; Wed, 13 Mar 2019 13:19:09 +0100 (CET) X-Virus-Scanned: amavisd-new at tudelft.nl Received: from mailservice.tudelft.nl ([130.161.131.73]) by localhost (tudelft.nl [127.0.0.1]) (amavisd-new, port 10026) with ESMTP id sWAng3iG6Z3p for ; Wed, 13 Mar 2019 13:19:07 +0100 (CET) Received: from smtp-a.tudelft.nl (smtp-a.tudelft.nl [131.180.190.158]) by mx2.tudelft.nl (Postfix) with ESMTP id 4370A6404F3 for ; Wed, 13 Mar 2019 13:19:07 +0100 (CET) Received: from mail-qk1-f176.google.com (mail-qk1-f176.google.com [209.85.222.176]) (using TLSv1.2 with cipher ECDHE-RSA-AES128-GCM-SHA256 (128/128 bits)) (No client certificate requested) by smtp-a.tudelft.nl (Postfix) with ESMTPSA id EE59820071 for ; Wed, 13 Mar 2019 13:19:06 +0100 (CET) Received: by mail-qk1-f176.google.com with SMTP id c1so906827qkk.4 for ; Wed, 13 Mar 2019 05:19:06 -0700 (PDT) X-Gm-Message-State: APjAAAU61Lav4w0m+RNPDDrp175gXj48nCUkgdt0612xdE96tCnHJ00h s6LcKDoJSr+4DABnhIm+0gWVWbRE/Q4KLR0dSXQ= X-Google-Smtp-Source: APXvYqwXuv3pgSp+fJGRvK2iar3f9jPAAPPiNh39OVAS7MMgVlOQAfq3WukItqqtNKnGGAsjZLN7OLY7LbOV5pUHNQc= X-Received: by 2002:a05:620a:1268:: with SMTP id b8mr14962774qkl.212.1552479546240; Wed, 13 Mar 2019 05:19:06 -0700 (PDT) MIME-Version: 1.0 References: <2e3749fa-c189-a756-2934-377510362cf8@apache.org> In-Reply-To: From: Wouter Zorgdrager Date: Wed, 13 Mar 2019 13:18:55 +0100 X-Gmail-Original-Message-ID: Message-ID: Subject: Re: Challenges using Flink REST API To: Chesnay Schepler Cc: "user@flink.apache.org" Content-Type: multipart/alternative; boundary="000000000000dff5bb0583f8cf04" --000000000000dff5bb0583f8cf04 Content-Type: text/plain; charset="UTF-8" Hey Chesnay, Actually I was mistaken by stating that in the JobManager logs I got the full stacktrace because I actually got the following there: 2019-03-13 11:55:13,906 ERROR org.apache.flink.runtime.webmonitor.handlers.JarRunHandler - Exception occurred in REST handler: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error. By some googling I came across this Jira issue [1], which seems to fix my issue in 1.8.0. However, I was still confused why this ever worked for me in 1.4.2 and by checking some binaries I found out that the REST API was reworked for 1.5.0 [2] which removed the full stack trace. Is there any (official) Docker image to already run Flink 1.8? Thanks, Wouter [1]: https://jira.apache.org/jira/browse/FLINK-11423 [2]: *https://jira.apache.org/jira/browse/FLINK-7715 * Op wo 13 mrt. 2019 om 12:18 schreef Chesnay Schepler : > Can you give me the stacktrace that is logged in the JobManager logs? > > > On 13.03.2019 10:57, Wouter Zorgdrager wrote: > > Hi Chesnay, > > Unfortunately this is not true when I run the Flink 1.7.2 docker images. > The response is still: > { > "errors": [ > "org.apache.flink.client.program.ProgramInvocationException: The > main method caused an error." > ] > } > > Regards, > Wouter Zorgdrager > > Op wo 13 mrt. 2019 om 10:42 schreef Chesnay Schepler : > >> You should get the full stacktrace if you upgrade to 1.7.2 . >> >> >> On 13.03.2019 09:55, Wouter Zorgdrager wrote: >> >> Hey all! >> >> I'm looking for some advice on the following; I'm working on an >> abstraction on top of Apache Flink to 'pipeline' Flink applications using >> Kafka. For deployment this means that all these Flink jobs are embedded >> into one jar and each job is started using an program argument (e.g. >> "--stage 'FirstFlinkJob'". To ease deploying a set of interconnected Flink >> jobs onto a cluster I wrote a Python script which basically communicates >> with the REST client of the JobManager. So you can do things like "pipeline >> start --jar 'JarWithThePipeline.jar'" and this would deploy every Flink >> application separately. >> >> However, this script was written a while ago against Flink version >> "1.4.2". This week I tried to upgrade it to Flink latest version but I >> noticed a change in the REST responses. In order to get the "pipeline >> start" command working,we need to know all the Flink jobs that are in the >> jar (we call these Flink jobs 'stages') because we need to know the stage >> names as argument for the jar. For the 1.4.2 version we used a dirty trick; >> we ran the jar with '--list --asException' as program arguments which >> basically runs the jar file and immediately throws an exception with the >> stage names. These are then parsed and used to start every stage >> separately. The error message that Flink threw looked something like this: >> >> java.util.concurrent.CompletionException: >> org.apache.flink.util.FlinkException: Could not run the jar. >> at >> org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleJsonRequest$0(JarRunHandler.java:90) >> at >> java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590) >> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) >> at java.util.concurrent.FutureTask.run(FutureTask.java:266) >> at >> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) >> at >> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) >> at >> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) >> at >> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) >> at java.lang.Thread.run(Thread.java:748) >> Caused by: org.apache.flink.util.FlinkException: Could not run the jar. >> ... 9 more >> Caused by: org.apache.flink.client.program.ProgramInvocationException: >> The main method caused an error. >> at >> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:542) >> at >> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:417) >> at >> org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(OptimizerPlanEnvironment.java:83) >> at >> org.apache.flink.client.program.ClusterClient.getOptimizedPlan(ClusterClient.java:334) >> at >> org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAndClassLoader(JarActionHandler.java:87) >> at >> org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleJsonRequest$0(JarRunHandler.java:69) >> ... 8 more >> Caused by: org.codefeedr.pipeline.PipelineListException: >> ["org.codefeedr.plugin.twitter.stages.TwitterStatusInput","mongo_tweets","elasticsearch_tweets"] >> at org.codefeedr.pipeline.Pipeline.showList(Pipeline.scala:114) >> at org.codefeedr.pipeline.Pipeline.start(Pipeline.scala:100) >> at nl.wouterr.Main$.main(Main.scala:23) >> at nl.wouterr.Main.main(Main.scala) >> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) >> at >> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) >> at >> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) >> at java.lang.reflect.Method.invoke(Method.java:498) >> at >> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:525) >> >> However, for 1.7.0 this trick doesn't work anymore because instead of >> returning the full stack trace, it only returns the following: >> org.apache.flink.client.program.ProgramInvocationException: The program >> caused an error: >> >> In the console of the JobManager it does give the full stack trace >> though. So first of all I'm wondering if there might be a way to enable >> more detailed stacktraces for Flink 1.7 in the REST responses. If not, do >> you have any suggestions on how to tackle this problem. I know, in the end >> this isn't really a Flink problem however you might know a workaround in >> the Flink REST client to achieve the same. >> >> Some solutions I already considered: >> - Running the jar with the "--list --asException" locally through the >> Python script; however Flink and Scala are not provided in the jar. >> Technically I could add them both to the classpath, but this would require >> users to have the Flink jar locally (and also Scala somewhere, but I assume >> most have). >> - Let users provide a list of stage names for all their (interconnected) >> Flink jobs. This is not really an option, because the (main) idea behind >> this framework is to reduce the boilerplate and cumbersome of setting up >> complex stream processing architectures. >> >> Any help is appreciated. Thanks in advance! >> >> Kind regards, >> Wouter Zorgdrager >> >> >> > --000000000000dff5bb0583f8cf04 Content-Type: text/html; charset="UTF-8" Content-Transfer-Encoding: quoted-printable
Hey Chesnay,

Actually I was mistaken by= stating that in the JobManager logs I got the full stacktrace because I ac= tually got the following there:
2019-03-13 11:55:13,906 ERRO= R org.apache.flink.runtime.webmonitor.handlers.JarRunHandler=C2=A0 =C2=A0 -= Exception occurred in REST handler: org.apache.flink.client.program.Progra= mInvocationException: The main method caused an error.

=
By some googling I came across this Jira issue [1], which seems = to fix my issue in 1.8.0. However, I was still confused why this ever worke= d for me in 1.4.2 and by checking some binaries I found out that the REST A= PI was reworked for 1.5.0 [2] which removed the full stack trace.

Is there any (official) Docker image to already run Flink 1= .8?

Thanks,
Wouter

<= div>[1]:=C2=A0https://jira.apache.org/jira/browse/FLINK-11423=C2=A0<= /div>


Op wo 13 mrt. 2019 om 12:18 schreef Chesnay Sche= pler <chesnay@ap= ache.org>:
=20 =20 =20
Can you give me the stacktrace that is logged in the JobManager logs?


On 13.03.2019 10:57, Wouter Zorgdrager wrote:
Hi Chesnay,

Unfortunately this is not true when I run the Flink 1.7.2 docker images. The response is still:
{
=C2=A0 =C2=A0 "errors": [
=C2=A0 =C2=A0 =C2=A0 =C2=A0 "org.apache.flink.client.program.ProgramInvocationExceptio= n: The main method caused an error."
=C2=A0 =C2=A0 ]
}

Regards,
Wouter Zorgdrager

Op wo 13 mrt. 2019 om 10:42 schreef Chesnay Schepler <chesnay@apache.org>:
You should get the full stacktrace if you upgrade to 1.7.2 .


On 13.03.2019 09:55, Wouter Zorgdrager wrote:
Hey all!

I'm looking for some advice on the following; I= 9;m working on an abstraction on top of Apache Flink to 'pipeline' Flink applications using Kafka. For deployment this means that all these Flink jobs are embedded into one jar and each job is started using an program argument (e.g. "--stage 'FirstFlinkJob&#= 39;". To ease deploying a set of interconnected Flink jobs onto a cluster I wrote a Python script which basically communicates with the REST client of the JobManager. So you can do things like "pipeline start --jar 'JarWithThePipeline.jar'" and this would dep= loy every Flink application separately.=C2=A0

However, this script was written a while ago against Flink version "1.4.2". This week I trie= d to upgrade it to Flink latest version but I noticed a change in the REST responses. In order to get the "pipeline start" command working,we need to kno= w all the Flink jobs that are in the jar (we call these Flink jobs 'stages') because we need to know the = stage names as argument for the jar. For the 1.4.2 version we used a dirty trick; we ran the jar with '--list --asException' as program arguments which basically runs the jar file and immediately throws an exception with the stage names. These are then parsed and used to start every stage separately. The error message that Flink threw looked something like this:

java.util.concurrent.Complet= ionException: org.apache.flink.util.FlinkException: Could not run the jar.
at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleJso= nRequest$0(JarRunHandler.java:90)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.ja= va:1590)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access= $201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Sc= heduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1= 149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:= 624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.util.FlinkException: Could not run the jar.
... 9 more
Caused by: org.apache.flink.client.program.ProgramInvocationExce= ption: The main method caused an error.
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProg= ram.java:542)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExe= cution(PackagedProgram.java:417)
at org.apache.flink.client.program.OptimizerPlanEnvironment.getOptimizedPlan(O= ptimizerPlanEnvironment.java:83)
at org.apache.flink.client.program.ClusterClient.getOptimizedPlan(ClusterClien= t.java:334)
at org.apache.flink.runtime.webmonitor.handlers.JarActionHandler.getJobGraphAn= dClassLoader(JarActionHandler.java:87)
at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleJso= nRequest$0(JarRunHandler.java:69)
... 8 more
Caused by: org.codefeedr.pipeline.PipelineListException: ["org.codefeedr.plugin.twitter.stages.TwitterStatusInput","m= ongo_tweets","elasticsearch_tweets"]
at org.codefeedr.pipeline.Pipeline.showList(Pipeline.scala:114)
at org.codefeedr.pipeline.Pipeline.start(Pipeline.scala:100)
at nl.wouterr.Main$.main(Main.scala:23)
at nl.wouterr.Main.main(Main.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:6= 2)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImp= l.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProg= ram.java:525)

However, for 1.7.0 this trick doesn't work anymore because instead of returning the full stack trace, it only returns the following:
org.apache.flink.client.progra= m.ProgramInvocationException: The program caused an error:=C2=A0

In the console of the JobManager it does give the full stack trace though. So first of all I'm wonderin= g if there might be a way to enable more detailed stacktraces for Flink 1.7 in the REST responses. If not, do you have any suggestions on how to tackle this problem. I know, in the end this isn't really a Flink problem however you might know a workaround in the Flink REST client to achieve the same.=C2=A0

Some solutions I already considered:
- Running the jar with the "--list --asException&= quot; locally through the Python script; however Flink and Scala are not provided in the jar. Technically I could add them both to the classpath, but this would require users to have the Flink jar locally (and also Scala somewhere, but I assume most have).=C2=A0=C2=A0
- Let users provide a list of stage names for all their (interconnected) Flink jobs. This is not really an option, because the (main) idea behind this framework is to reduce the boilerplate and cumbersome of setting up complex stream processing architectures.=C2=A0

Any help is appreciated. Thanks in advance!

Kind regards,
Wouter Zorgdrager



--000000000000dff5bb0583f8cf04--