Skip to content

Commit 8d30256

Browse files
shangeyaocursoragent
shangeyao
and
cursoragent
authored
[Migrate] Migrate streampark-flink-packer from Scala to Java (#4463)
* [Migrate] #4449 Migrate streampark-flink-packer from Scala to Java Replace all 34 Scala sources in the packer module with Java equivalents, preserving public API for console and flink-client callers. Tests migrated to JUnit 5 with Surefire workspace configuration for Maven resolve tests. Co-authored-by: Cursor <cursoragent@cursor.com> * [CI] Fix Sonar reliability issues for #4463 Handle InterruptedException in BuildPipeline.launch and check uberJar.delete() result in MavenTool. Co-authored-by: Cursor <cursoragent@cursor.com> * [CI] Fix DockerClientTest and Sonar reliability issues for #4463 Rebase onto dev after #4462 merge; prepare hello-world test image in CI, replace Throwable catches with specific exception handling, and document intentionally empty silent watcher callbacks. Co-authored-by: Cursor <cursoragent@cursor.com> * [CI] Fix Scala/Java Map interop after packer migration for #4463 DockerImageBuildResponse.podTemplatePaths() now returns a Java Map; convert via JavaConverters before iterating in KubernetesNativeClientTrait. Co-authored-by: Cursor <cursoragent@cursor.com> * [Build] Use properties for packer dependency versions in #4463 Replace hard-coded jackson-annotations and httpclient versions with property references aligned to the root POM. Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent be3229e commit 8d30256

91 files changed

Lines changed: 5739 additions & 3352 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/KubernetesNativeClientTrait.scala‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,8 @@ import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.Service
3333

3434
import javax.annotation.Nonnull
3535

36+
import scala.collection.JavaConverters.mapAsScalaMapConverter
37+
3638
/** kubernetes native mode submit */
3739
trait KubernetesNativeClientTrait extends FlinkClientTrait {
3840

@@ -48,7 +50,7 @@ trait KubernetesNativeClientTrait extends FlinkClientTrait {
4850
if (submitRequest.buildResult != null && submitRequest.deployMode == FlinkDeployMode.KUBERNETES_NATIVE_APPLICATION) {
4951
val buildResult =
5052
submitRequest.buildResult.asInstanceOf[DockerImageBuildResponse]
51-
buildResult.podTemplatePaths.foreach(p => {
53+
buildResult.podTemplatePaths.asScala.foreach(p => {
5254
if (PodTemplateTool.KUBERNETES_POD_TEMPLATE.key.equals(p._1)) {
5355
flinkConfig.safeSet(KubernetesConfigOptions.KUBERNETES_POD_TEMPLATE, p._2)
5456
} else if (PodTemplateTool.KUBERNETES_JM_POD_TEMPLATE.key.equals(p._1)) {

‎streampark-flink/streampark-flink-packer/pom.xml‎

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,9 +30,9 @@
3030
<properties>
3131
<eclipse.aether.version>1.1.0</eclipse.aether.version>
3232
<maven.aether.version>3.3.9</maven.aether.version>
33-
<maven.shade.version>3.2.4</maven.shade.version>
33+
<maven.shade.version>${maven-shade-plugin.version}</maven.shade.version>
3434
<docker.client.version>3.3.6</docker.client.version>
35-
<scalatest.skiptests>true</scalatest.skiptests>
35+
<httpclient.version>4.5.14</httpclient.version>
3636
</properties>
3737

3838
<dependencies>
@@ -140,10 +140,31 @@
140140
<dependency>
141141
<groupId>com.fasterxml.jackson.core</groupId>
142142
<artifactId>jackson-annotations</artifactId>
143-
<version>2.12.0</version>
143+
<version>${jackson.version}</version>
144+
</dependency>
145+
146+
<dependency>
147+
<groupId>org.apache.httpcomponents</groupId>
148+
<artifactId>httpclient</artifactId>
149+
<version>${httpclient.version}</version>
150+
<scope>test</scope>
144151
</dependency>
145152
</dependencies>
146153

154+
<build>
155+
<plugins>
156+
<plugin>
157+
<groupId>org.apache.maven.plugins</groupId>
158+
<artifactId>maven-surefire-plugin</artifactId>
159+
<configuration>
160+
<systemPropertyVariables>
161+
<streampark.workspace.local>${project.build.directory}/streampark-workspace</streampark.workspace.local>
162+
</systemPropertyVariables>
163+
</configuration>
164+
</plugin>
165+
</plugins>
166+
</build>
167+
147168
<profiles>
148169
<profile>
149170
<id>apache-release</id>

‎streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/PackerResourceGC.java‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.packer;
19+
20+
import org.apache.streampark.common.conf.Workspace;
21+
import org.apache.streampark.common.constants.Constants;
22+
import org.apache.streampark.common.util.LoggerSupport;
23+
24+
import org.apache.commons.io.FileUtils;
25+
26+
import java.io.File;
27+
import java.util.ArrayList;
28+
import java.util.Arrays;
29+
import java.util.List;
30+
31+
/** Garbage resource collector during packing. */
32+
public final class PackerResourceGC extends LoggerSupport {
33+
34+
private static final PackerResourceGC INSTANCE = new PackerResourceGC();
35+
36+
private static final String APP_WORKSPACE_PATH = Workspace.local().APP_WORKSPACE();
37+
38+
private PackerResourceGC() {
39+
}
40+
41+
/** Start a building legacy resources collection process. */
42+
public static void startGc(Integer expiredHours) {
43+
INSTANCE.doStartGc(expiredHours);
44+
}
45+
46+
private void doStartGc(Integer expiredHours) {
47+
File appWorkspace = new File(APP_WORKSPACE_PATH);
48+
if (!appWorkspace.exists()) {
49+
return;
50+
}
51+
long evictedBarrier = System.currentTimeMillis() - expiredHours * 3600L * 1000L;
52+
53+
File[] dirs = appWorkspace.listFiles(File::isDirectory);
54+
if (dirs == null) {
55+
return;
56+
}
57+
List<File> evictedFiles = new ArrayList<>();
58+
for (File dir : dirs) {
59+
if (!dir.getName().contains("@")) {
60+
continue;
61+
}
62+
for (FileWithTime entry : findLastModifiedOfSubFile(dir)) {
63+
if (entry.lastModified < evictedBarrier) {
64+
evictedFiles.add(entry.file);
65+
}
66+
}
67+
}
68+
69+
if (evictedFiles.isEmpty()) {
70+
return;
71+
}
72+
StringBuilder sb = new StringBuilder();
73+
for (File path : evictedFiles) {
74+
if (sb.length() > 0) {
75+
sb.append(", ");
76+
}
77+
sb.append(path.getAbsolutePath());
78+
}
79+
logInfo("Delete expired building resources, " + sb);
80+
for (File path : evictedFiles) {
81+
try {
82+
FileUtils.deleteDirectory(path);
83+
} catch (Exception ignored) {
84+
// ignore delete failures
85+
}
86+
}
87+
}
88+
89+
private static List<FileWithTime> findLastModifiedOfSubFile(File file) {
90+
File[] children = file.listFiles();
91+
if (children == null) {
92+
return new ArrayList<>();
93+
}
94+
boolean isApplicationMode =
95+
Arrays.stream(children).anyMatch(f -> f.getName().contains(Constants.JAR_SUFFIX));
96+
List<FileWithTime> result = new ArrayList<>();
97+
if (isApplicationMode) {
98+
long max = Arrays.stream(children).mapToLong(File::lastModified).max().orElse(0L);
99+
result.add(new FileWithTime(file, max));
100+
} else {
101+
for (File subFile : children) {
102+
if (!subFile.isDirectory()) {
103+
continue;
104+
}
105+
File[] subChildren = subFile.listFiles();
106+
if (subChildren == null) {
107+
continue;
108+
}
109+
long max = Arrays.stream(subChildren).mapToLong(File::lastModified).max().orElse(0L);
110+
result.add(new FileWithTime(subFile, max));
111+
}
112+
}
113+
return result;
114+
}
115+
116+
private static final class FileWithTime {
117+
118+
private final File file;
119+
private final long lastModified;
120+
121+
private FileWithTime(File file, long lastModified) {
122+
this.file = file;
123+
this.lastModified = lastModified;
124+
}
125+
}
126+
}

‎streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/docker/DockerConf.java‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.packer.docker;
19+
20+
import com.github.dockerjava.api.model.AuthConfig;
21+
22+
import javax.annotation.Nullable;
23+
24+
/** Authentication Configuration of Remote Docker Register */
25+
public final class DockerConf {
26+
27+
@Nullable
28+
private final String registerAddress;
29+
private final String imageNamespace;
30+
private final String registerUsername;
31+
private final String registerPassword;
32+
33+
public DockerConf(
34+
@Nullable String registerAddress,
35+
String imageNamespace,
36+
String registerUsername,
37+
String registerPassword) {
38+
this.registerAddress = registerAddress;
39+
this.imageNamespace = imageNamespace;
40+
this.registerUsername = registerUsername;
41+
this.registerPassword = registerPassword;
42+
}
43+
44+
@Nullable
45+
public String registerAddress() {
46+
return registerAddress;
47+
}
48+
49+
public String imageNamespace() {
50+
return imageNamespace;
51+
}
52+
53+
public String registerUsername() {
54+
return registerUsername;
55+
}
56+
57+
public String registerPassword() {
58+
return registerPassword;
59+
}
60+
61+
/** covert to com.github.docker.java.api.model.AuthConfig */
62+
public AuthConfig toAuthConf() {
63+
return new AuthConfig()
64+
.withRegistryAddress(registerAddress)
65+
.withUsername(registerUsername)
66+
.withPassword(registerPassword);
67+
}
68+
69+
public static DockerConf of(
70+
@Nullable String registerAddress,
71+
String imageNameSpace,
72+
String registerUsername,
73+
String registerPassword) {
74+
return new DockerConf(registerAddress, imageNameSpace, registerUsername, registerPassword);
75+
}
76+
}

‎streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/docker/DockerImageExist.java‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.packer.docker;
19+
20+
import com.github.dockerjava.api.exception.NotFoundException;
21+
22+
/** Check docker image existence. */
23+
public class DockerImageExist {
24+
25+
public boolean doesDockerImageExist(String imageName) {
26+
return DockerUtils.usingDockerClient(
27+
dockerClient -> {
28+
try {
29+
dockerClient.inspectImageCmd(imageName).exec();
30+
return true;
31+
} catch (NotFoundException e) {
32+
return false;
33+
}
34+
},
35+
err -> {
36+
throw new RuntimeException(
37+
"Check docker image failed, imageName=" + imageName, err);
38+
});
39+
}
40+
}

‎streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/docker/DockerRetriever.java‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.packer.docker;
19+
20+
import org.apache.streampark.common.conf.CommonConfig;
21+
import org.apache.streampark.common.conf.InternalConfigHolder;
22+
import org.apache.streampark.common.util.Utils;
23+
24+
import com.github.dockerjava.api.DockerClient;
25+
import com.github.dockerjava.core.DefaultDockerClientConfig;
26+
import com.github.dockerjava.core.DockerClientConfig;
27+
import com.github.dockerjava.core.HackDockerClient;
28+
import com.github.dockerjava.httpclient5.ApacheDockerHttpClient;
29+
30+
import java.net.URI;
31+
import java.time.Duration;
32+
33+
/** Docker client factory. */
34+
public final class DockerRetriever {
35+
36+
public static final DockerClientConfig dockerClientConf =
37+
DefaultDockerClientConfig.createDefaultConfigBuilder().build();
38+
39+
private static final ApacheDockerHttpClient.Builder dockerHttpClientBuilder =
40+
new ApacheDockerHttpClient.Builder()
41+
.dockerHost(dockerClientConf.getDockerHost())
42+
.sslConfig(dockerClientConf.getSSLConfig())
43+
.maxConnections(InternalConfigHolder.get(CommonConfig.DOCKER_MAX_CONNECTIONS()))
44+
.connectionTimeout(
45+
Duration.ofSeconds(
46+
InternalConfigHolder.get(CommonConfig.DOCKER_CONNECTION_TIMEOUT_SEC())))
47+
.responseTimeout(
48+
Duration.ofSeconds(
49+
InternalConfigHolder.get(CommonConfig.DOCKER_RESPONSE_TIMEOUT_SEC())));
50+
51+
private DockerRetriever() {
52+
}
53+
54+
/** get new DockerClient instance */
55+
public static DockerClient newDockerClient() {
56+
setDockerHost();
57+
return HackDockerClient.getInstance(dockerClientConf, dockerHttpClientBuilder.build());
58+
}
59+
60+
/** set docker-host for kata */
61+
private static void setDockerHost() {
62+
String dockerHost = InternalConfigHolder.get(CommonConfig.DOCKER_HOST());
63+
if (Utils.isNotEmpty(dockerHost)) {
64+
URI dockerHostUri = URI.create(dockerHost);
65+
dockerHttpClientBuilder.dockerHost(dockerHostUri);
66+
}
67+
}
68+
}

0 commit comments

Comments
 (0)