wolfboys commented on code in PR #3668:
URL:
https://github.com/apache/incubator-streampark/pull/3668#discussion_r1565641719
##########
streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/KubernetesRetriever.scala:
##########
@@ -89,15 +89,19 @@ object KubernetesRetriever extends Logger {
val clusterProvider: KubernetesClusterDescriptor =
clientFactory.createClusterDescriptor(flinkConfig).asInstanceOf[KubernetesClusterDescriptor]
- Try {
- clusterProvider
- .retrieve(flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID))
- .getClusterClient
- } match {
- case Success(v) => Some(v)
- case Failure(e) =>
- logError(s"Get flinkClient error, the error is: $e")
- None
+ try {
Review Comment:
Can you test the improved code below? If it can pass the test, it is
recommended to use the following code
```scala
// retrieve flink cluster client
clusterClientServiceLoader
.getClusterClientFactory(flinkConfig)
.createClusterDescriptor(flinkConfig)
.asInstanceOf[KubernetesClusterDescriptor]
.autoClose {
clusterProvider =>
Try {
clusterProvider
.retrieve(flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID))
.getClusterClient
} match {
case Success(v) => Some(v)
case Failure(e) =>
logError(s"Get flinkClient error, the error is: $e")
None
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@streampark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
