You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/07/05 03:19:25 UTC
[GitHub] [flink] link3280 opened a new pull request, #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
link3280 opened a new pull request, #20159:
URL: https://github.com/apache/flink/pull/20159
## What is the purpose of the change
Support stop job statement in SQL client. Subtask of FLIP-222.
## Brief change log
- Support stop-job statement in Flink SQL parser
- Support stop-job operation in SQL Client and LocalExecutor
## Verifying this change
This change added tests and can be verified as follows:
- Added tests to verify stop-job statement in CliClientTest
- Added tests to verify stop-job operation in LocalExecutorITCase
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduces a new feature? yes
- If yes, how is the feature documented? docs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1174564013
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fba1490c2c59d4c3b817b6e1420d3caac7fe0492",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fba1490c2c59d4c3b817b6e1420d3caac7fe0492",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fba1490c2c59d4c3b817b6e1420d3caac7fe0492 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r924595473
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,96 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
Review Comment:
I agree that there are some duplicate codes between `CliFrontEnd` and `LocalExecutor`, but the method signatures are different. `CliFrontEnd` gets the cluster ID from the command line and local configurations, while `LocalExecutor` gets the cluster ID from the session configuration. Given the duplicated codes are few, I think it's acceptable. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r924224755
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,96 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
Review Comment:
LGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r928106727
##########
flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl:
##########
@@ -2296,3 +2296,38 @@ SqlNode SqlAnalyzeTable():
return new SqlAnalyzeTable(s.end(this), tableName, partitionSpec, columns, allColumns);
}
}
+
+/**
+* Parses a STOP JOB statement:
+* STOP JOB <JOB_ID> [<WITH SAVEPOINT>] [<WITH DRAIN>];
+*/
+SqlStopJob SqlStopJob() :
+{
+ SqlCharStringLiteral jobId;
+ boolean isWithSavepoint = false;
+ boolean isWithDrain = false;
+}
+{
+ <STOP> <JOB> <QUOTED_STRING>
+ {
+ String id = SqlParserUtil.parseString(token.image);
+ jobId = SqlLiteral.createCharString(id, getPos());
+ }
+ [
+ LOOKAHEAD(2)
+ <WITH> <SAVEPOINT>
Review Comment:
I've reconsidered the syntax of `WITH DRAIN`, and found `WITH DRAIN` should be an attribute to `WITH SAVEPOINT`. That's to say, we could not have `WITH DRAIN` without `WITH SAVEPOINT`.
How about we keep the order and validate that `WITH DRAIN` could only appear after `WITH SAVEPOINT`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1190954677
> Thanks for your contribution. Currently the `SqlGatewayService` already supported executeSql, could you also support this feature in the SQL Gateway?
@fsk119 `SqlGatewayService` is under rapid development, could we make it as a follow-up PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r924230945
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java:
##########
@@ -406,6 +420,38 @@ public void testBatchQueryExecutionMultipleTimes() throws Exception {
}
}
+ @Test(timeout = 90_000L)
+ public void testStopJob() throws Exception {
+ final Map<String, String> configMap = new HashMap<>();
+ configMap.put(EXECUTION_RESULT_MODE.key(), ResultMode.TABLE.name());
+ configMap.put(RUNTIME_MODE.key(), RuntimeExecutionMode.STREAMING.name());
+ configMap.put(TableConfigOptions.TABLE_DML_SYNC.key(), "false");
+
+ final LocalExecutor executor =
+ createLocalExecutor(
+ Collections.singletonList(udfDependency), Configuration.fromMap(configMap));
+ String sessionId = executor.openSession("test-session");
+
+ final String srcDdl = "CREATE TABLE src (a STRING) WITH ('connector' = 'datagen')";
+ final String snkDdl = "CREATE TABLE snk (a STRING) WITH ('connector' = 'blackhole')";
+ final String insert = "INSERT INTO snk SELECT a FROM src;";
+
+ try {
+ executor.executeOperation(sessionId, executor.parseStatement(sessionId, srcDdl));
+ executor.executeOperation(sessionId, executor.parseStatement(sessionId, snkDdl));
+ TableResult result =
+ executor.executeOperation(
+ sessionId, executor.parseStatement(sessionId, insert));
+ JobID jobId = result.getJobClient().get().getJobID();
+ // wait till the job turns into running status
+ Thread.sleep(2_000L);
Review Comment:
Good point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1190280802
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
fsk119 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r924018891
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/tcl/SqlStopJob.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.sql.parser.tcl;
+
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.NlsString;
+
+import javax.annotation.Nonnull;
+
+import java.util.Collections;
+import java.util.List;
+
+/** Stop job command to stop a flink job. */
Review Comment:
How about:
```
The command to stop a flink job.
```
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,96 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
+ String sessionId, ClusterAction<ClusterID, Result> clusterAction)
+ throws FlinkException {
+ final SessionContext context = getSessionContext(sessionId);
+ final Configuration configuration = (Configuration) context.getReadableConfig();
+ final ClusterClientFactory<ClusterID> clusterClientFactory =
+ clusterClientServiceLoader.getClusterClientFactory(configuration);
+
+ final ClusterID clusterId = clusterClientFactory.getClusterId(configuration);
+ Preconditions.checkNotNull(clusterId, "No cluster ID found for session " + sessionId);
+
+ try (final ClusterDescriptor<ClusterID> clusterDescriptor =
+ clusterClientFactory.createClusterDescriptor(configuration)) {
+ try (final ClusterClient<ClusterID> clusterClient =
+ clusterDescriptor.retrieve(clusterId).getClusterClient()) {
+ return clusterAction.runAction(clusterClient);
+ }
+ }
Review Comment:
Simplify:
```
try (final ClusterDescriptor<ClusterID> clusterDescriptor =
clusterClientFactory.createClusterDescriptor(configuration);
final ClusterClient<ClusterID> clusterClient =
clusterDescriptor.retrieve(clusterId).getClusterClient()) {
return clusterAction.runAction(clusterClient);
}
```
##########
flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java:
##########
@@ -1683,6 +1683,11 @@ void testTryCast() {
"TRY_CAST(`A` AS ROW(`F0` INTEGER ARRAY, `F1` MAP< STRING, DECIMAL(10, 2) >, `F2` STRING NOT NULL))");
}
+ @Test
+ void testStopJob() {
+ sql("STOP JOB 'myjob'").ok("STOP JOB 'myjob'");
Review Comment:
Also, check `with savepoint` and `with drain`.
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java:
##########
@@ -115,6 +121,14 @@ private static Configuration getConfig() {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TMS);
config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, NUM_SLOTS_PER_TM);
config.setBoolean(WebOptions.SUBMIT_ENABLE, false);
+ config.set(StateBackendOptions.STATE_BACKEND, "hashmap");
+ config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
+ config.set(
+ CheckpointingOptions.CHECKPOINTS_DIRECTORY,
+ Paths.get(System.getProperty("java.io.tmpdir")).toUri().toString());
+ config.set(
+ CheckpointingOptions.SAVEPOINT_DIRECTORY,
+ Paths.get(System.getProperty("java.io.tmpdir")).toUri().toString());
Review Comment:
Why not set the temporary folder path?
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java:
##########
@@ -406,6 +420,38 @@ public void testBatchQueryExecutionMultipleTimes() throws Exception {
}
}
+ @Test(timeout = 90_000L)
+ public void testStopJob() throws Exception {
+ final Map<String, String> configMap = new HashMap<>();
+ configMap.put(EXECUTION_RESULT_MODE.key(), ResultMode.TABLE.name());
+ configMap.put(RUNTIME_MODE.key(), RuntimeExecutionMode.STREAMING.name());
+ configMap.put(TableConfigOptions.TABLE_DML_SYNC.key(), "false");
+
+ final LocalExecutor executor =
+ createLocalExecutor(
+ Collections.singletonList(udfDependency), Configuration.fromMap(configMap));
+ String sessionId = executor.openSession("test-session");
+
+ final String srcDdl = "CREATE TABLE src (a STRING) WITH ('connector' = 'datagen')";
+ final String snkDdl = "CREATE TABLE snk (a STRING) WITH ('connector' = 'blackhole')";
+ final String insert = "INSERT INTO snk SELECT a FROM src;";
+
+ try {
+ executor.executeOperation(sessionId, executor.parseStatement(sessionId, srcDdl));
+ executor.executeOperation(sessionId, executor.parseStatement(sessionId, snkDdl));
+ TableResult result =
+ executor.executeOperation(
+ sessionId, executor.parseStatement(sessionId, insert));
+ JobID jobId = result.getJobClient().get().getJobID();
+ // wait till the job turns into running status
+ Thread.sleep(2_000L);
Review Comment:
We can use the job client to fetch the job status.
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/tcl/SqlStopJob.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.sql.parser.tcl;
Review Comment:
TCL means transaction control language. Actually, we don't offer any transaction-level semantics here. From my side, I think it's more like a DDL, which manages metadata of the running jobs. What do you think?
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliClientTest.java:
##########
@@ -329,6 +332,68 @@ public void testCancelExecutionInteractiveMode() throws Exception {
}
}
+ @Test(timeout = 10000)
+ public void testStopJob() throws Exception {
+ final MockExecutor mockExecutor = new MockExecutor();
+ mockExecutor.isSync = false;
+
+ String sessionId = mockExecutor.openSession("test-session");
+ OutputStream outputStream = new ByteArrayOutputStream(256);
+ try (CliClient client =
+ new CliClient(
+ () -> TerminalUtils.createDumbTerminal(outputStream),
+ sessionId,
+ mockExecutor,
+ historyTempFile(),
+ null)) {
+ client.executeInNonInteractiveMode(INSERT_INTO_STATEMENT);
+ String dmlResult = outputStream.toString();
+
+ Pattern pattern = Pattern.compile("[\\s\\S]*Job ID: (.*)[\\s\\S]*");
+ Matcher matcher = pattern.matcher(dmlResult);
+ assertThat(matcher.matches()).isTrue();
+ String jobId = matcher.group(1);
Review Comment:
Add a method, e.g. getJobID to reuse these codes.
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,96 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
Review Comment:
It seems we plan to copy most of the codes from the `CliFrontEnd` about ClusterAction. I just think it's better to make the `ClusterAction` public and we can reuse these codes. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1174575436
cc @wuchong @fsk119 @godfreyhe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 merged pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
fsk119 merged PR #20159:
URL: https://github.com/apache/flink/pull/20159
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
fsk119 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r927344410
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java:
##########
@@ -115,6 +121,14 @@ private static Configuration getConfig() {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TMS);
config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, NUM_SLOTS_PER_TM);
config.setBoolean(WebOptions.SUBMIT_ENABLE, false);
+ config.set(StateBackendOptions.STATE_BACKEND, "hashmap");
+ config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
+ config.set(
+ CheckpointingOptions.CHECKPOINTS_DIRECTORY,
+ Paths.get(System.getProperty("java.io.tmpdir")).toUri().toString());
+ config.set(
+ CheckpointingOptions.SAVEPOINT_DIRECTORY,
+ Paths.get(System.getProperty("java.io.tmpdir")).toUri().toString());
Review Comment:
I think we have two ways to solve this:
1. we can use junit5 api, which has `@Order` annotation to control the order. You can cc SqlGatewayStatemetnITCase.
2. we can set this value in the test
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/context/SessionContext.java:
##########
@@ -118,6 +118,10 @@ Set<URL> getDependencies() {
return dependencies;
}
+ public URLClassLoader getClassLoader() {
+ return classLoader;
+ }
Review Comment:
It seems no one uses this. I think we can remove it.
##########
flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl:
##########
@@ -2296,3 +2296,38 @@ SqlNode SqlAnalyzeTable():
return new SqlAnalyzeTable(s.end(this), tableName, partitionSpec, columns, allColumns);
}
}
+
+/**
+* Parses a STOP JOB statement:
+* STOP JOB <JOB_ID> [<WITH SAVEPOINT>] [<WITH DRAIN>];
+*/
+SqlStopJob SqlStopJob() :
+{
+ SqlCharStringLiteral jobId;
+ boolean isWithSavepoint = false;
+ boolean isWithDrain = false;
+}
+{
+ <STOP> <JOB> <QUOTED_STRING>
+ {
+ String id = SqlParserUtil.parseString(token.image);
+ jobId = SqlLiteral.createCharString(id, getPos());
+ }
+ [
+ LOOKAHEAD(2)
+ <WITH> <SAVEPOINT>
Review Comment:
It seems it's impossible to parse
```
STOP JOB 'myjob' WITH DRAIN WITH SAVEPOINT
```
I think we can do as `EXPLAIN` syntax that allows users to specify `ExplainDetail` in any order.
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,95 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
+ String sessionId, ClusterAction<ClusterID, Result> clusterAction)
+ throws FlinkException {
+ final SessionContext context = getSessionContext(sessionId);
+ final Configuration configuration = (Configuration) context.getReadableConfig();
+ final ClusterClientFactory<ClusterID> clusterClientFactory =
+ clusterClientServiceLoader.getClusterClientFactory(configuration);
+
+ final ClusterID clusterId = clusterClientFactory.getClusterId(configuration);
Review Comment:
Actually `ClusterClientServiceLoader` is not thread-safe. It uses the
```
final ServiceLoader<ClusterClientFactory> loader =
ServiceLoader.load(ClusterClientFactory.class);
```
to load the Factory, which uses the thread-level classloader. It's better we can add a method named `getClusterId` in the `SessionContext`. I think it brings benefits:
1. we don't cast the ReadableConfig to Configuration
2. we can use the session-level classloader inside the session context.
With these benefits, we only need to pay the cost: create the `ClusterServiceLoader` per session.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1174881715
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
fsk119 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r925095048
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,96 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
Review Comment:
Okay. We can refactor when neeeds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r924224755
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,96 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
Review Comment:
LGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r928108479
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,95 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
+ String sessionId, ClusterAction<ClusterID, Result> clusterAction)
+ throws FlinkException {
+ final SessionContext context = getSessionContext(sessionId);
+ final Configuration configuration = (Configuration) context.getReadableConfig();
+ final ClusterClientFactory<ClusterID> clusterClientFactory =
+ clusterClientServiceLoader.getClusterClientFactory(configuration);
+
+ final ClusterID clusterId = clusterClientFactory.getClusterId(configuration);
Review Comment:
But only `ClusterID` is not enough, we also need `ClusterDescriptor` which also comes from `ClusterClientFactory`, thus I think we need to add a method called `createClusterDescriptor` to `SessionContext` as well. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r924218611
##########
flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/tcl/SqlStopJob.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.sql.parser.tcl;
Review Comment:
Good point. DDL LGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fsk119 commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
fsk119 commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1190968837
> > Thanks for your contribution. Currently the `SqlGatewayService` already supported executeSql, could you also support this feature in the SQL Gateway?
>
> @fsk119 `SqlGatewayService` is under rapid development, could we make it as a follow-up PR?
Ok. Let move it to the next PR. Could you open a ticket for this? I will review again today.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1221708160
ping @fsk119
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r924226857
##########
flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java:
##########
@@ -115,6 +121,14 @@ private static Configuration getConfig() {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TMS);
config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, NUM_SLOTS_PER_TM);
config.setBoolean(WebOptions.SUBMIT_ENABLE, false);
+ config.set(StateBackendOptions.STATE_BACKEND, "hashmap");
+ config.set(CheckpointingOptions.CHECKPOINT_STORAGE, "filesystem");
+ config.set(
+ CheckpointingOptions.CHECKPOINTS_DIRECTORY,
+ Paths.get(System.getProperty("java.io.tmpdir")).toUri().toString());
+ config.set(
+ CheckpointingOptions.SAVEPOINT_DIRECTORY,
+ Paths.get(System.getProperty("java.io.tmpdir")).toUri().toString());
Review Comment:
Cause temporary folders could not be used in static blocks, as they're not initialized yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on a diff in pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on code in PR #20159:
URL: https://github.com/apache/flink/pull/20159#discussion_r928108479
##########
flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/LocalExecutor.java:
##########
@@ -322,4 +339,95 @@ public List<String> listJars(String sessionId) {
final SessionContext context = getSessionContext(sessionId);
return context.listJars();
}
+
+ @Override
+ public Optional<String> stopJob(
+ String sessionId, String jobId, boolean isWithSavepoint, boolean isWithDrain)
+ throws SqlExecutionException {
+ Duration clientTimeout = getSessionConfig(sessionId).get(ClientOptions.CLIENT_TIMEOUT);
+ try {
+ return runClusterAction(
+ sessionId,
+ clusterClient -> {
+ if (isWithSavepoint) {
+ // blocking get savepoint path
+ try {
+ String savepoint =
+ clusterClient
+ .stopWithSavepoint(
+ JobID.fromHexString(jobId),
+ isWithDrain,
+ null,
+ SavepointFormatType.DEFAULT)
+ .get(
+ clientTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
+ return Optional.of(savepoint);
+ } catch (Exception e) {
+ throw new FlinkException(
+ "Could not stop job "
+ + jobId
+ + " in session "
+ + sessionId
+ + ".",
+ e);
+ }
+ } else {
+ clusterClient.cancel(JobID.fromHexString(jobId));
+ return Optional.empty();
+ }
+ });
+ } catch (Exception e) {
+ throw new SqlExecutionException(
+ "Could not stop job " + jobId + " in session " + sessionId + ".", e);
+ }
+ }
+
+ /**
+ * Retrieves the {@link ClusterClient} from the session and runs the given {@link ClusterAction}
+ * against it.
+ *
+ * @param sessionId the specified session ID
+ * @param clusterAction the cluster action to run against the retrieved {@link ClusterClient}.
+ * @param <ClusterID> type of the cluster id
+ * @param <Result>> type of the result
+ * @throws FlinkException if something goes wrong
+ */
+ private <ClusterID, Result> Result runClusterAction(
+ String sessionId, ClusterAction<ClusterID, Result> clusterAction)
+ throws FlinkException {
+ final SessionContext context = getSessionContext(sessionId);
+ final Configuration configuration = (Configuration) context.getReadableConfig();
+ final ClusterClientFactory<ClusterID> clusterClientFactory =
+ clusterClientServiceLoader.getClusterClientFactory(configuration);
+
+ final ClusterID clusterId = clusterClientFactory.getClusterId(configuration);
Review Comment:
The only problem is that `ClusterID` is not enough, we also need `ClusterDescriptor<ClusterID>` which also comes from `ClusterClientFactory<ClusterID>`, that requries SessionContext to know about <ClusterID> .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1190976770
New ticket https://issues.apache.org/jira/browse/FLINK-28617.
I'll open tickets/PRs for the gateway separately for the rest statements.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] link3280 commented on pull request #20159: [FLINK-28360][SQL Client] Support stop job statement in SQL client
Posted by GitBox <gi...@apache.org>.
link3280 commented on PR #20159:
URL: https://github.com/apache/flink/pull/20159#issuecomment-1174881511
The build failed due to an unrelated test https://issues.apache.org/jira/browse/FLINK-28391.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org