Recommended approach to debug this
Locked
12
12
Re: Recommended approach to debug this
 
|
Well, I think I got the solution though I am not yet sure of the problem .. The original code looked like this .. Try { // from a parent class called Runner which runs a streamlet // run returns an abstraction which completes a Promise depending on whether // the Job was successful or not val streamletExecution = loadedStreamlet.streamlet.run(withPodRuntimeConfig) // In normal circumstances it will run forever for streaming data source unless // being stopped forcibly or any of the queries faces an exception Await.result(streamletExecution.completed, Duration.Inf) } match { //.. and then the streamlet.run(..) in turn finally invoked the following .. val env = StreamExecutionEnvironment.getExecutionEnvironment // creates datastreams and read from / writes to Kafka // I pasted the body of this earlier in the thread buildExecutionGraph() env.execute(..) This DID NOT run and failed with the exception I reported earlier. But when I change the code to get the run statement out of the Try block, things run fine .. like this .. // from a parent class called Runner which runs a streamlet // run returns an abstraction which completes a Promise depending on whether // the Job was successful or not val streamletExecution = loadedStreamlet.streamlet.run(withPodRuntimeConfig) Try { // the runner waits for the execution to complete// In normal circumstances it will run forever for streaming data source unless // being stopped forcibly or any of the queries faces an exception Await.result(streamletExecution.completed, Duration.Inf) } match { //.. Apparently it looks like the exception that I was facing earlier leaked through the Flink engine and Try caught it and it got logged. But removing it out of Try now enables Flink to catch it back and follow the course that it should. But I am not sure if this is a cogent explanation and looking forward to some more accurate one from the experts. Note there is no asynchrony of concurrency going on here - the Runner code may look a bit over-engineered but there is a context to this. The Runner code handles not only Flink but other types of streaming engines as well like Spark and Akka Streams. regards. On Tue, Sep 24, 2019 at 10:17 AM Biao Liu <[hidden email]> wrote:
Debasish Ghosh http://manning.com/ghosh2 |
|
|
The key point of this case is in `PackagedProgram#callMainMethod`. The `ProgramAbortException` is expected when executing the main method here. This `ProgramAbortException` thrown is wrapped with `InvocationTargetException` by Java reflection layer [1]. There is a piece of codes handling `InvocationTargetException`. try { mainMethod.invoke(null, (Object) args); } catch (... catch (InvocationTargetException e) { Throwable exceptionInMethod = e.getTargetException(); if (exceptionInMethod instanceof Error) { throw (Error) exceptionInMethod; ------> `ProgramAbortException` would be caught expectedly here. } else if (exceptionInMethod instanceof ProgramParametrizationException) { throw (ProgramParametrizationException) exceptionInMethod; } else if (exceptionInMethod instanceof ProgramInvocationException) { throw (ProgramInvocationException) exceptionInMethod; } else { ------> If I'm right, the wrapped exception (Boxed Error or something else) change the exception, it is caught here throw new ProgramInvocationException("The main method caused an error: " + exceptionInMethod.getMessage(), exceptionInMethod); } The `ProgramInvocationException` is handled specially in `OptimizerPlanEnvironment`. try { prog.invokeInteractiveModeForExecution(); } catch (ProgramInvocationException e) { throw e; ------> The submission is failed here in this case } catch (Throwable t) { // the invocation gets aborted with the preview plan if (optimizerPlan != null) { return optimizerPlan; ------> Normally it should be here } else { throw new ProgramInvocationException("The program caused an error: ", t); } ... Thanks, Biao /'bɪ.aʊ/ On Tue, 24 Sep 2019 at 17:35, Debasish Ghosh <[hidden email]> wrote:
|
|
|
So I believe (I did't test it) the solution for this case is keeping the original exception thrown from `env.execute()` and throwing this exception out of main method. It's a bit tricky, maybe we could have a better design of this scenario. Thanks, Biao /'bɪ.aʊ/ On Tue, 24 Sep 2019 at 18:55, Biao Liu <[hidden email]> wrote:
|
|
|
Actually there is an ongoing client API refactoring on this stuff[1] and one of the main purpose is eliminating hijacking env.execute... Biao Liu <[hidden email]> 于2019年9月24日周二 下午7:12写道：
|
|
|
Hi Zili, Great to hear that! Hope to see the new client soon! Thanks, Biao /'bɪ.aʊ/ On Tue, 24 Sep 2019 at 19:23, Zili Chen <[hidden email]> wrote:
|
«
Return to (DEPRECATED) Apache Flink User Mailing List archive.
|
1 view|%1 views
| Free forum by Nabble | Edit this page |