Re: [PR] [FLINK-35194][table] Support describe job with job id [flink]
lsyldliu closed pull request #24728: [FLINK-35194][table] Support describe job with job id URL: https://github.com/apache/flink/pull/24728 -- 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35194][table] Support describe job with job id [flink]
xuyangzhong commented on PR #24728: URL: https://github.com/apache/flink/pull/24728#issuecomment-2084076648 The doc jira is: https://issues.apache.org/jira/browse/FLINK-35271 -- 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [FLINK-35194][table] Support describe job with job id [flink]
xuyangzhong commented on code in PR #24728:
URL: https://github.com/apache/flink/pull/24728#discussion_r1583035157
##
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##
@@ -511,6 +511,57 @@ void testShowJobsOperation(@InjectClusterClient
RestClusterClient restCluster
.isBetween(timeOpStart, timeOpSucceed);
}
+@Test
+void testDescribeJobOperation(@InjectClusterClient RestClusterClient
restClusterClient)
+throws Exception {
+SessionHandle sessionHandle =
service.openSession(defaultSessionEnvironment);
+Configuration configuration = new
Configuration(MINI_CLUSTER.getClientConfiguration());
+
+String pipelineName = "test-describe-job";
+configuration.set(PipelineOptions.NAME, pipelineName);
+
+// running jobs
+String sourceDdl = "CREATE TABLE source (a STRING) WITH
('connector'='datagen');";
+String sinkDdl = "CREATE TABLE sink (a STRING) WITH
('connector'='blackhole');";
+String insertSql = "INSERT INTO sink SELECT * FROM source;";
+
+service.executeStatement(sessionHandle, sourceDdl, -1, configuration);
+service.executeStatement(sessionHandle, sinkDdl, -1, configuration);
+
+long timeOpStart = System.currentTimeMillis();
+OperationHandle insertsOperationHandle =
+service.executeStatement(sessionHandle, insertSql, -1,
configuration);
+String jobId =
+fetchAllResults(sessionHandle, insertsOperationHandle)
+.get(0)
+.getString(0)
+.toString();
+
+TestUtils.waitUntilAllTasksAreRunning(restClusterClient,
JobID.fromHexString(jobId));
+long timeOpSucceed = System.currentTimeMillis();
+
+OperationHandle describeJobOperationHandle =
+service.executeStatement(
+sessionHandle,
+String.format("DESCRIBE JOB '%s'", jobId),
+-1,
+configuration);
+
+List result = fetchAllResults(sessionHandle,
describeJobOperationHandle);
+RowData jobRow =
+result.stream()
+.filter(row ->
jobId.equals(row.getString(0).toString()))
+.findFirst()
+.orElseThrow(
+() ->
+new IllegalStateException(
+"Test job " + jobId + " not
found."));
+assertThat(jobRow.getString(1)).hasToString(pipelineName);
+assertThat(jobRow.getString(2)).hasToString("RUNNING");
+assertThat(jobRow.getTimestamp(3, 3).getMillisecond())
+.isBetween(timeOpStart, timeOpSucceed);
+}
Review Comment:
No, all jobs will be cancelled automatically. You can find more details at
`MiniClusterExtension#afterAll`.
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [FLINK-35194][table] Support describe job with job id [flink]
xuyangzhong commented on code in PR #24728:
URL: https://github.com/apache/flink/pull/24728#discussion_r1583029959
##
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##
@@ -511,6 +511,57 @@ void testShowJobsOperation(@InjectClusterClient
RestClusterClient restCluster
.isBetween(timeOpStart, timeOpSucceed);
}
+@Test
+void testDescribeJobOperation(@InjectClusterClient RestClusterClient
restClusterClient)
+throws Exception {
+SessionHandle sessionHandle =
service.openSession(defaultSessionEnvironment);
+Configuration configuration = new
Configuration(MINI_CLUSTER.getClientConfiguration());
+
+String pipelineName = "test-describe-job";
+configuration.set(PipelineOptions.NAME, pipelineName);
+
+// running jobs
+String sourceDdl = "CREATE TABLE source (a STRING) WITH
('connector'='datagen');";
+String sinkDdl = "CREATE TABLE sink (a STRING) WITH
('connector'='blackhole');";
+String insertSql = "INSERT INTO sink SELECT * FROM source;";
+
+service.executeStatement(sessionHandle, sourceDdl, -1, configuration);
+service.executeStatement(sessionHandle, sinkDdl, -1, configuration);
+
+long timeOpStart = System.currentTimeMillis();
+OperationHandle insertsOperationHandle =
+service.executeStatement(sessionHandle, insertSql, -1,
configuration);
+String jobId =
+fetchAllResults(sessionHandle, insertsOperationHandle)
+.get(0)
+.getString(0)
+.toString();
+
+TestUtils.waitUntilAllTasksAreRunning(restClusterClient,
JobID.fromHexString(jobId));
+long timeOpSucceed = System.currentTimeMillis();
+
+OperationHandle describeJobOperationHandle =
+service.executeStatement(
+sessionHandle,
+String.format("DESCRIBE JOB '%s'", jobId),
+-1,
+configuration);
+
+List result = fetchAllResults(sessionHandle,
describeJobOperationHandle);
+RowData jobRow =
+result.stream()
+.filter(row ->
jobId.equals(row.getString(0).toString()))
+.findFirst()
+.orElseThrow(
+() ->
+new IllegalStateException(
+"Test job " + jobId + " not
found."));
+assertThat(jobRow.getString(1)).hasToString(pipelineName);
+assertThat(jobRow.getString(2)).hasToString("RUNNING");
Review Comment:
Yes, it's always RUNNING, because `ExecutionOptions.RUNTIME_MODE` is
streaming default.
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [FLINK-35194][table] Support describe job with job id [flink]
lsyldliu commented on code in PR #24728:
URL: https://github.com/apache/flink/pull/24728#discussion_r1582558688
##
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java:
##
@@ -774,6 +777,53 @@ public ResultFetcher callShowJobsOperation(
resultRows);
}
+public ResultFetcher callDescribeJobOperation(
+TableEnvironmentInternal tableEnv,
+OperationHandle operationHandle,
+DescribeJobOperation describeJobOperation)
+throws SqlExecutionException {
+Configuration configuration = tableEnv.getConfig().getConfiguration();
+Duration clientTimeout =
configuration.get(ClientOptions.CLIENT_TIMEOUT);
+String jobId = describeJobOperation.getJobId();
+Optional jobStatusOp =
+runClusterAction(
+configuration,
+operationHandle,
+clusterClient -> {
+try {
+JobID expectedJobId =
JobID.fromHexString(jobId);
+return clusterClient.listJobs()
+.get(clientTimeout.toMillis(),
TimeUnit.MILLISECONDS)
+.stream()
+.filter(job ->
expectedJobId.equals(job.getJobId()))
+.findFirst();
+} catch (Exception e) {
+throw new SqlExecutionException(
+"Failed to get jobs in the cluster.",
e);
Review Comment:
String.format("Failed to get job %s in the cluster.", jobId)
##
flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java:
##
@@ -774,6 +777,53 @@ public ResultFetcher callShowJobsOperation(
resultRows);
}
+public ResultFetcher callDescribeJobOperation(
+TableEnvironmentInternal tableEnv,
+OperationHandle operationHandle,
+DescribeJobOperation describeJobOperation)
+throws SqlExecutionException {
+Configuration configuration = tableEnv.getConfig().getConfiguration();
+Duration clientTimeout =
configuration.get(ClientOptions.CLIENT_TIMEOUT);
+String jobId = describeJobOperation.getJobId();
+Optional jobStatusOp =
+runClusterAction(
+configuration,
+operationHandle,
+clusterClient -> {
+try {
+JobID expectedJobId =
JobID.fromHexString(jobId);
+return clusterClient.listJobs()
+.get(clientTimeout.toMillis(),
TimeUnit.MILLISECONDS)
+.stream()
+.filter(job ->
expectedJobId.equals(job.getJobId()))
+.findFirst();
+} catch (Exception e) {
+throw new SqlExecutionException(
+"Failed to get jobs in the cluster.",
e);
+}
+});
+
+if (!jobStatusOp.isPresent()) {
+throw new SqlExecutionException("The job described by " + jobId +
" does not exist.");
Review Comment:
String.format("Described job %s does not exist in the cluster.", jobId)?
##
flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/SqlGatewayServiceITCase.java:
##
@@ -511,6 +511,57 @@ void testShowJobsOperation(@InjectClusterClient
RestClusterClient restCluster
.isBetween(timeOpStart, timeOpSucceed);
}
+@Test
+void testDescribeJobOperation(@InjectClusterClient RestClusterClient
restClusterClient)
+throws Exception {
+SessionHandle sessionHandle =
service.openSession(defaultSessionEnvironment);
+Configuration configuration = new
Configuration(MINI_CLUSTER.getClientConfiguration());
+
+String pipelineName = "test-describe-job";
+configuration.set(PipelineOptions.NAME, pipelineName);
+
+// running jobs
+String sourceDdl = "CREATE TABLE source (a STRING) WITH
('connector'='datagen');";
+String sinkDdl = "CREATE TABLE sink (a STRING) WITH
('connector'='blackhole');";
+String insertSql = "INSERT INTO sink SELECT * FROM source;";
+
+service.executeStatement(sessionHandle, sourceDdl, -1, configuration);
+service.executeStatement(sessionHandle, sinkDdl, -1, configuration);
+
+long timeOpStart = System.currentTimeMillis();
+OperationHandle insertsOperationHandle =
+
Re: [PR] [FLINK-35194][table] Support describe job with job id [flink]
flinkbot commented on PR #24728: URL: https://github.com/apache/flink/pull/24728#issuecomment-2079305845 ## CI report: * 6511c8930cde9f8a3e3acef78fc3bcecb5e9cb51 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run azure` re-run the last Azure build -- 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-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
