[Bug] [Connector] Use Console sink will submit two job in flink cluster

[Hisoka-X]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

When use console sink, and submit job to flink cluster, client will sumbit two job.

SeaTunnel Version

dev

SeaTunnel Config

env {
  execution.parallelism = 2
}

source {

  JdbcSource {
    driver = "com.mysql.cj.jdbc.Driver"
    url = "jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf8&useSSL=false"
    username = "root"
    password = "123456"
    query = "SELECT int_type,char_type,bigint_type from test_test where 1=1"
    result_table_name = "user"
    partition_column = "int_type"
    partition_upper_bound = 50
    partition_lower_bound = 0
    parallelism = 4
  }

}

transform {

}
sink {
  ConsoleSink {}
}

Running Command

./bin/start-seatunnel-flink.sh --config ./config/flink.batch.conf

Error Exception

-

Flink or Spark Version

Flink 1.13.6

Java or Scala Version

Java8

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[ruanwenjun]

Yes, I also find this, and this will happen in batch job.

[ruanwenjun]

@benjfan I find this may be caused by we have added two DataSink in ConsoleSink? https://github.com/apache/incubator-seatunnel/blob/3a4ef4c60d8397fbf7ee7b16a06494072d3f0245/seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-console/src/main/java/org/apache/seatunnel/flink/console/sink/ConsoleSink.java#L47-L54

After change to below code, this bug has been fixed.

   @Override
    public DataSink<Row> outputBatch(FlinkEnvironment env, DataSet<Row> rowDataSet) {
        return rowDataSet.output(this);
    }

   @SuppressWarnings("RegexpSingleline")
    @Override
    public void writeRecord(Row record) {
        if (limit <= 0) {
            return;
        }
        System.out.println(record.toString());
        limit--;
    }

[Hisoka-X]

@benjfan I find this may be caused by we have added two DataSink in ConsoleSink?

https://github.com/apache/incubator-seatunnel/blob/3a4ef4c60d8397fbf7ee7b16a06494072d3f0245/seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-console/src/main/java/org/apache/seatunnel/flink/console/sink/ConsoleSink.java#L47-L54

After change to below code, this bug has been fixed.

   @Override
    public DataSink<Row> outputBatch(FlinkEnvironment env, DataSet<Row> rowDataSet) {
        return rowDataSet.output(this);
    }

   @SuppressWarnings("RegexpSingleline")
    @Override
    public void writeRecord(Row record) {
        if (limit <= 0) {
            return;
        }
        System.out.println(record.toString());
        limit--;
    }

This method will make record print on TaskManager, not client side.

[ruanwenjun]

@benjfan I find this may be caused by we have added two DataSink in ConsoleSink?
https://github.com/apache/incubator-seatunnel/blob/3a4ef4c60d8397fbf7ee7b16a06494072d3f0245/seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-console/src/main/java/org/apache/seatunnel/flink/console/sink/ConsoleSink.java#L47-L54

After change to below code, this bug has been fixed.

   @Override
    public DataSink<Row> outputBatch(FlinkEnvironment env, DataSet<Row> rowDataSet) {
        return rowDataSet.output(this);
    }

   @SuppressWarnings("RegexpSingleline")
    @Override
    public void writeRecord(Row record) {
        if (limit <= 0) {
            return;
        }
        System.out.println(record.toString());
        limit--;
    }

This method will make record print on TaskManager, not client side.

Yes, you are right, we cannot easily do this change. The main reason is that Flink will do execute in collect method, and this will conflict with our execute.