[FLINK-15669] SQL client can't cancel flink job

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.10.0
• Fix Version/s: 1.10.1, 1.11.0
• Component/s: Table SQL / Client
• Labels: pull-request-available
• Assignee: godfrey he
• Reporter: godfrey he
• Created: 19/Jan/20 11:36
• Updated: 27/Apr/20 09:22
• Resolved: 27/Apr/20 09:22

Description

in sql client, CLI client do cancel query operation through void cancelQuery(String sessionId, String resultId) method in Executor. However, the resultId is a random UUID, is not the job id. So CLI client can't cancel a running job.

related code in LocalExecutor:

private <C> ResultDescriptor executeQueryInternal(String sessionId, ExecutionContext<C> context, String query) {
	 ......

	// store the result with a unique id
	final String resultId = UUID.randomUUID().toString();
	resultStore.storeResult(resultId, result);

	......

	// create execution
	final ProgramDeployer deployer = new ProgramDeployer(
		configuration, jobName, pipeline);

	// start result retrieval
	result.startRetrieval(deployer);

	return new ResultDescriptor(
			resultId,
			removeTimeAttributes(table.getSchema()),
			result.isMaterialized());
}



private <T> void cancelQueryInternal(ExecutionContext<T> context, String resultId) {
	......

	// stop Flink job
	try (final ClusterDescriptor<T> clusterDescriptor = context.createClusterDescriptor()) {
		ClusterClient<T> clusterClient = null;
		try {
			// retrieve existing cluster
			clusterClient = clusterDescriptor.retrieve(context.getClusterId()).getClusterClient();
			try {
				// ======== cancel job through resultId =======
				clusterClient.cancel(new JobID(StringUtils.hexStringToByte(resultId))).get();
			} catch (Throwable t) {
				// the job might has finished earlier
			}
		} catch (Exception e) {
			throw new SqlExecutionException("Could not retrieve or create a cluster.", e);
		} finally {
			try {
				if (clusterClient != null) {
					clusterClient.close();
				}
			} catch (Exception e) {
				// ignore
			}
		}
	} catch (SqlExecutionException e) {
		throw e;
	} catch (Exception e) {
		throw new SqlExecutionException("Could not locate a cluster.", e);
	}
}

Issue Links

• relates to: FLINK-17405 add test cases for cancel job in SQL client
• links to: GitHub Pull Request #10904