[FLINK-18545] Sql api cannot specify flink job name

• Type: Improvement
• Status: Closed
• Priority: Critical
• Resolution: Done
• Affects Version/s: 1.11.0
• Fix Version/s: 1.11.3, 1.12.0
• Component/s: Client / Job Submission, Table SQL / API
• Labels: pull-request-available

Environment

execute sql : 

StreamTableEnvironment.executeSql("insert into user_log_sink select user_id, item_id, category_id, behavior, ts from user_log")

current job name :  org.apache.flink.table.api.internal.TableEnvironmentImpl

public TableResult executeInternal(List<ModifyOperation> operations) {
    List<Transformation<?>> transformations = translate(operations);
    List<String> sinkIdentifierNames = extractSinkIdentifierNames(operations);
    String jobName = "insert-into_" + String.join(",", sinkIdentifierNames);
    Pipeline pipeline = execEnv.createPipeline(transformations, tableConfig, jobName);
    try {
        JobClient jobClient = execEnv.executeAsync(pipeline);
        TableSchema.Builder builder = TableSchema.builder();
        Object[] affectedRowCounts = new Long[operations.size()];
        for (int i = 0; i < operations.size(); ++i) {
            // use sink identifier name as field name
            builder.field(sinkIdentifierNames.get(i), DataTypes.BIGINT());
            affectedRowCounts[i] = -1L;
        }        return TableResultImpl.builder()
                .jobClient(jobClient)
                .resultKind(ResultKind.SUCCESS_WITH_CONTENT)
                .tableSchema(builder.build())
                .data(Collections.singletonList(Row.of(affectedRowCounts)))
                .build();
    } catch (Exception e) {
        throw new TableException("Failed to execute sql", e);
    }
}

• Release Note: Introduce `pipeline.name` to allow users to specify job name by configuration. This option does not break existing pipelines.
• Assignee: godfrey he
• Reporter: venn wu
• Created: 09/Jul/20 13:06
• Updated: 19/May/21 06:08
• Resolved: 27/Nov/20 06:58

Description

In Flink 1.11.0, StreamTableEnvironment.executeSql(sql) will explan and execute job Immediately, The job name will special as "insert-into_sink-table-name".  But we have Multiple sql job will insert into a same sink table, this is not very friendly. 

Issue Links

• duplicates: FLINK-22707 StatementSet support exectue(jobName)
• duplicates: FLINK-20242 Support to set job name in StatementSet.execute() method
• duplicates: FLINK-16709 add a set command to set job name when submit job on sql client
• links to: GitHub Pull Request #13215
• links to: GitHub Pull Request #14103
• links to: GitHub Pull Request #14233