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<JobStatusMessage> 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<JobStatusMessage> 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 =
+ 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<RowData> 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:
Do need we cancel the job manually after getting the job status?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/SqlNodeToOperationConversion.java:
##########
@@ -1273,6 +1277,10 @@ private Operation convertShowJobs(SqlShowJobs
sqlStopJob) {
return new ShowJobsOperation();
}
+ private Operation convertDescribeJob(SqlDescribeJob sqlDescribeJob) {
Review Comment:
It would be better we could migrate this convert logic to `SqlNodeConverter`.
##########
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<RowData> 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:
The job status always be`RUNNING` here? maybe it has finished? My concern is
whether it causes an unstable test?
--
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
