[Bug] When the task was submitted, it failed due to version number issues.

cgxCode commented 1 year ago

Search before asking

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

Java Version

openjdk version "1.8.0_232"

Scala Version

2.12.x

StreamPark Version

2.12-2.1.0

Flink Version

The CDH version of Flink is 1.14.0-csa1.7.0.0.

deploy mode

yarn-application

What happened

I want to integrate the CDH version number into StreamPark 2.1.0. However, when I submit a job, I receive an error.

I tried to modify the data in the database to make it compile successfully, but when I executed it, it still threw an error.

I think we need to retrieve the version number at the bottom layer of the interface and convert it to a number, but the Flink version of CDH is quite special.

Error Exception

java.util.concurrent.CompletionException: java.lang.NumberFormatException: For input string: "0-csa1"
    at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
    at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
    at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1606)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.NumberFormatException: For input string: "0-csa1"
    at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
    at java.lang.Integer.parseInt(Integer.java:580)
    at java.lang.Integer.parseInt(Integer.java:615)
    at scala.collection.immutable.StringLike.toInt(StringLike.scala:304)
    at scala.collection.immutable.StringLike.toInt$(StringLike.scala:304)
    at scala.collection.immutable.StringOps.toInt(StringOps.scala:33)
    at org.apache.streampark.common.conf.FlinkVersion.$anonfun$checkVersion$1(FlinkVersion.scala:117)
    at org.apache.streampark.common.conf.FlinkVersion.$anonfun$checkVersion$1$adapted(FlinkVersion.scala:117)
    at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:237)
    at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)
    at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)
    at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:198)
    at scala.collection.TraversableLike.map(TraversableLike.scala:237)
    at scala.collection.TraversableLike.map$(TraversableLike.scala:230)
    at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:198)
    at org.apache.streampark.common.conf.FlinkVersion.checkVersion(FlinkVersion.scala:117)
    at org.apache.streampark.flink.client.FlinkClient$.proxy(FlinkClient.scala:72)
    at org.apache.streampark.flink.client.FlinkClient$.submit(FlinkClient.scala:49)
    at org.apache.streampark.flink.client.FlinkClient.submit(FlinkClient.scala)
    at org.apache.streampark.console.core.service.impl.ApplicationServiceImpl.lambda$start$10(ApplicationServiceImpl.java:1544)
    at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
    ... 3 more

Screenshots

No response

Are you willing to submit PR?

[ ] Yes I am willing to submit a PR!(您是否要贡献这个PR?)

Code of Conduct

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

cgxCode commented 1 year ago

Streampark supports only standard version of Flink and does not support CDH/CDP version of Flink client. Even if you fork the source code to bypass the version check mentioned above, there will be more incompatibility issues that you will face later. The biggest problem is that many parameter settings in Streampark are set using "-Dkey=value", but it is not effective in CDH/CDP version of Flink. streampark支持的是flink标准版本，不支持cdh/cdp版本的flink客户端。即使你fork了源码将上述版本校验绕过去，后面会有更多的不兼容。最大的问题在于：streampark的很多参数设置都是通过-Dkey=value进行设置，然而cdp的flink这样设置是无效的。

wolfboys commented 4 months ago

fixed, version >= 2.1.2

apache / incubator-streampark