[Bug] [seatunnel-api-flink] Can not find BatchTableEnvironment class when using Flink-1.15.X

[s7monk]

Search before asking

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

What happened

Can not find BatchTableEnvironment class when using Flink-1.15.X

SeaTunnel Version

dev

SeaTunnel Config

env {
  # You can set flink configuration here
  execution.parallelism = 1
  #execution.checkpoint.interval = 10000
  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
}

source {
  # This is a example source plugin **only for test and demonstrate the feature source plugin**
    Jdbc {
      driver = com.mysql.cj.jdbc.Driver
      url = "jdbc:mysql://rm-bp17bvb1v77ad16l1to.mysql.rds.aliyuncs.com:3306/data_warehouse?serverTimezone=GMT%2b8&useSSL=false"
      user = "root"
      password = "123456"
      query = "select * from user"
      result_table_name = "test"
  }

  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
  # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
}

transform {

  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
}

sink {
  Console{}

  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
  # please go to https://seatunnel.apache.org/docs/flink/configuration/sink-plugins/Console
}

Running Command

bin/start-seatunnel-flink-connector-v2.sh -c job/rds_to_console.conf

Error Exception

java.lang.NoClassDefFoundError: org/apache/flink/table/api/bridge/java/BatchTableEnvironment
	at org.apache.seatunnel.flink.FlinkEnvironment.createBatchTableEnvironment(FlinkEnvironment.java:229)
	at org.apache.seatunnel.flink.FlinkEnvironment.prepare(FlinkEnvironment.java:96)
	at org.apache.seatunnel.flink.FlinkEnvironment.prepare(FlinkEnvironment.java:55)
	at org.apache.seatunnel.core.starter.config.EnvironmentFactory.getEnvironment(EnvironmentFactory.java:46)
	at org.apache.seatunnel.core.starter.flink.execution.FlinkExecution.<init>(FlinkExecution.java:55)
	at org.apache.seatunnel.core.starter.flink.command.FlinkApiTaskExecuteCommand.execute(FlinkApiTaskExecuteCommand.java:53)
	at org.apache.seatunnel.core.starter.Seatunnel.run(Seatunnel.java:40)
	at org.apache.seatunnel.core.starter.flink.SeatunnelFlink.main(SeatunnelFlink.java:34)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
	at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:836)
	at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:247)
	at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1078)
	at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1156)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1729)
	at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
	at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1156)
Caused by: java.lang.ClassNotFoundException: org.apache.flink.table.api.bridge.java.BatchTableEnvironment
	at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:68)
	at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65)
	at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:52)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
	... 24 more

Flink or Spark Version

Flink-1.15.2

Java or Scala Version

java8

Screenshots

No response

Are you willing to submit PR?

• [X] Yes I am willing to submit a PR!

Code of Conduct

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

[EricJoy2048]

Great.

[hk-lrzy]

Maybe i can work on this one if you don't have ant free time. @s7monk

[liugddx]

This can be added to the roadmap? @EricJoy2048

[github-actions[bot]]

This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs.

[liugddx]

#3263

[github-actions[bot]]

This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs.