[FLINK-14840] Use Executor interface in SQL cli #10313

aljoscha · 2019-11-25T12:50:40Z

What is the purpose of the change

Change the SQL cli to use the new Executor interface and JobClient instead of manually going through cluster descriptor/cluster specification.

This also does not print the cluster ID and web interface URL anymore
when submitting because there is no portable way of retrieving these.
Previously, even for YARN the printed cluster id was not usable because
it was just the class name of the implementation class.

Verifying this change

This change is covered by existing tests and ITcases that were adapted.

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

flinkbot · 2019-11-25T12:53:01Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 64339da (Wed Dec 04 17:08:48 UTC 2019)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2019-11-25T13:15:17Z

CI report:

939f98f : FAILURE Build
4a5ca62 : SUCCESS Build
ab81c5b : FAILURE Build
ec7ba22 : SUCCESS Build
1a774b9 : SUCCESS Build
730cdb5 : CANCELED Build
7c52d1c : CANCELED Build
bc8a2de : FAILURE Build
64339da : FAILURE Build
c10cc10 : SUCCESS Build
bd147a0 : SUCCESS Build
464d8d4 : CANCELED Build
b50be35 : CANCELED Build
588f931 : SUCCESS Build
170f6e3 : CANCELED Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

...-sql-client/src/main/java/org/apache/flink/table/client/gateway/ProgramTargetDescriptor.java

kl0u

Nice work @aljoscha !
I had some comments with the main ones being the one about not closing the jobClient and the one about changing the threading model of the MaterializedCollectBatchResult.

...nt/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectStreamResult.java

.../java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java

...k-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java

...link-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java

...-sql-client/src/main/java/org/apache/flink/table/client/gateway/ProgramTargetDescriptor.java

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java

tisonkun

Thanks for opening this pull request @aljoscha ! I left several comments while response to @zjffdu 's & @kl0u 's comments also. Please ping me when you address them :-)

...-sql-client/src/main/java/org/apache/flink/table/client/gateway/ProgramTargetDescriptor.java

...k-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java

...link-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java

...nt/src/main/java/org/apache/flink/table/client/gateway/local/result/CollectStreamResult.java

.../java/org/apache/flink/table/client/gateway/local/result/MaterializedCollectBatchResult.java

twalthr · 2019-11-26T08:29:06Z

Thanks for the PR @aljoscha. I would like to take a final look before merging. Please ping me when there are no more concerns.

aljoscha · 2019-11-26T13:04:03Z

I think I addressed all your comments. Please take another look.

aljoscha · 2019-11-27T15:04:55Z

Is everyone OK with me rebasing and squashing this so that Timo can have a look?

tisonkun

No more concerns from my side. Go ahead :-)

kl0u

@aljoscha I did my pass and I left some comments mainly having to do with style rather than essence. After integrating whichever you find meaningful, you have my +1, but I would feel more comfortable if someone who also knows more about the SQL Client, like @twalthr, has a look before merging, as I am not that familiar with this part of the codebase.

kl0u · 2019-11-28T14:00:54Z

...k-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java

+			CommandLine commandLine,
+			Options commandLineOptions,
+			List<CustomCommandLine> availableCommandLines) throws FlinkException {
+


This options is only used for logging. Do we need it or we can simply print the final executionConfig ?

kl0u · 2019-11-28T14:03:09Z

...k-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java

-		}
-
-		private Pipeline createPipeline(String name, Configuration flinkConfig) {
+		public Pipeline createPipeline(String name, Configuration flinkConfig) {


In this method it seems we do not use the flinkConfig or the parallelism in the else branch. Is it ok to remove them? Also the else is not needed as we return in the if branch. So this can become:

public Pipeline createPipeline(String name) { if (streamExecEnv != null) { // special case for Blink planner to apply batch optimizations // note: it also modifies the ExecutionConfig! if (executor instanceof ExecutorBase) { return ((ExecutorBase) executor).getStreamGraph(name); } return streamExecEnv.getStreamGraph(name); } return execEnv.createProgramPlan(name); }

kl0u · 2019-11-28T14:03:41Z

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java

 			boolean awaitJobResult) {
 		this.context = context;
-		this.jobGraph = jobGraph;
+		this.pipeline = pipeline;


checkNotNull?

kl0u · 2019-11-28T14:10:44Z

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java

 	 * @param awaitJobResult block for a job execution result from the cluster
 	 */
 	public ProgramDeployer(
 			ExecutionContext<C> context,


Also here, for core readability, I think it is easier to pass the configuration directly instead of creating in the deploy and when we fetch the result, have a checkState on the ATTACHED flag so that the reader of the code knows what to expect. The way it is now, it seems like in the deploy, we are setting the ATTACHED flag but it is never used.

WDYT?

kl0u · 2019-11-28T14:11:37Z

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java

-			throw e;
-		} catch (Exception e) {
-			throw new SqlExecutionException("Could not locate a cluster.", e);
+		if (awaitJobResult) {


What about configuration.set(DeploymentOptions.ATTACHED, awaitJobResult); ?

kl0u · 2019-11-28T14:12:12Z

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java

-			} catch (Exception e) {
-				// ignore
-			}
+			jobClient = executor.execute(pipeline, configuration);


Why not directly return executor.execute(pipeline, configuration); ?

aljoscha · 2019-12-05T10:04:19Z

@flinkbot run travis

This also does not print the cluster ID and web interface URL anymore when submitting because there is no portable way of retrieving these. Previously, even for YARN the printed cluster id was not usable because it was just the class name of the implementation class.

aljoscha requested review from kl0u and tisonkun November 25, 2019 12:50

rmetzger added the review=description? label Nov 25, 2019

rmetzger added the component=TableSQL/Client label Nov 25, 2019

aljoscha force-pushed the executor-sql-cli-without-job-client-changes branch 2 times, most recently from df63633 to 4a5ca62 Compare November 25, 2019 13:48

zjffdu reviewed Nov 25, 2019

View reviewed changes

...-sql-client/src/main/java/org/apache/flink/table/client/gateway/ProgramTargetDescriptor.java Outdated Show resolved Hide resolved

kl0u requested changes Nov 25, 2019

View reviewed changes

zjffdu reviewed Nov 26, 2019

View reviewed changes

...nk-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ProgramDeployer.java Outdated Show resolved Hide resolved

tisonkun requested changes Nov 26, 2019

View reviewed changes

tisonkun approved these changes Nov 27, 2019

View reviewed changes

kl0u approved these changes Nov 28, 2019

View reviewed changes

aljoscha force-pushed the executor-sql-cli-without-job-client-changes branch 6 times, most recently from 64339da to c10cc10 Compare December 5, 2019 09:22

aljoscha force-pushed the executor-sql-cli-without-job-client-changes branch 3 times, most recently from b50be35 to 588f931 Compare December 6, 2019 22:05

aljoscha added 2 commits December 7, 2019 17:03

[FLINK-14840] Add unit tests for exception behaviour of SQL cli executor

de1169b

aljoscha force-pushed the executor-sql-cli-without-job-client-changes branch from 588f931 to 170f6e3 Compare December 7, 2019 16:15

aljoscha closed this Dec 7, 2019

aljoscha deleted the executor-sql-cli-without-job-client-changes branch December 7, 2019 22:43

[FLINK-14840] Use Executor interface in SQL cli #10313

[FLINK-14840] Use Executor interface in SQL cli #10313

Uh oh!

Conversation

aljoscha commented Nov 25, 2019

What is the purpose of the change

Verifying this change

Documentation

Uh oh!

flinkbot commented Nov 25, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Automated Checks

Review Progress

Uh oh!

flinkbot commented Nov 25, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Uh oh!

kl0u left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tisonkun left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

twalthr commented Nov 26, 2019

Uh oh!

aljoscha commented Nov 26, 2019

Uh oh!

aljoscha commented Nov 27, 2019

Uh oh!

tisonkun left a comment

Choose a reason for hiding this comment

Uh oh!

kl0u left a comment

Choose a reason for hiding this comment

Uh oh!

kl0u Nov 28, 2019

Choose a reason for hiding this comment

Uh oh!

kl0u Nov 28, 2019

Choose a reason for hiding this comment

Uh oh!

kl0u Nov 28, 2019

Choose a reason for hiding this comment

Uh oh!

kl0u Nov 28, 2019

Choose a reason for hiding this comment

Uh oh!

kl0u Nov 28, 2019

Choose a reason for hiding this comment

Uh oh!

kl0u Nov 28, 2019

Choose a reason for hiding this comment

Uh oh!

aljoscha commented Dec 5, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

flinkbot commented Nov 25, 2019 •

edited

Loading

flinkbot commented Nov 25, 2019 •

edited

Loading