originalUris) {
+ return originalUris.stream().anyMatch(uri -> uri.contains("local:/"));
+ }
+}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/artifact/KubernetesArtifactUploader.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/artifact/KubernetesArtifactUploader.java
new file mode 100644
index 0000000000000..16580b82a0235
--- /dev/null
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/artifact/KubernetesArtifactUploader.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.artifact;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.client.cli.ArtifactFetchOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+
+/** Local artifact uploader for Kubernetes programs. */
+@Internal
+public interface KubernetesArtifactUploader {
+
+ /**
+ * Uploads all {@code local://} schemed artifact that is present, according to the given
+ * configuration. Any remote artifact remains as it was passed originally.
+ *
+ * Takes the job JAR from the {@link PipelineOptions#JARS} config and any additional
+ * artifacts from the {@link ArtifactFetchOptions#ARTIFACT_LIST} config. After the upload,
+ * replaces the URIs of any local JAR in these configs to point to the remotely available one.
+ *
+ *
Requires {@link KubernetesConfigOptions#LOCAL_UPLOAD_TARGET} to point to a valid, existing
+ * DFS directory with read and write permissions.
+ *
+ * @param config given Flink configuration
+ * @throws Exception when the upload process fails
+ */
+ void uploadAll(Configuration config) throws Exception;
+}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java
index a2e2e7c9b054e..88fb3fcfe3f05 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java
@@ -537,6 +537,26 @@ public class KubernetesConfigOptions {
+ "Flink. A typical use-case is when one uses Flink Kubernetes "
+ "Operator.");
+ public static final ConfigOption LOCAL_UPLOAD_ENABLED =
+ ConfigOptions.key("kubernetes.artifacts.local-upload-enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Enables uploading 'local://' schemed artifacts to DFS before the the application cluster deployment.");
+
+ public static final ConfigOption LOCAL_UPLOAD_OVERWRITE =
+ ConfigOptions.key("kubernetes.artifacts.local-upload-overwrite")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "If enabled, overwrites any existing artifact on the remote target. Disabled by default.");
+
+ public static final ConfigOption LOCAL_UPLOAD_TARGET =
+ ConfigOptions.key("kubernetes.artifacts.local-upload-target")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The target remote DFS directory to upload local artifacts.");
+
/**
* This will only be used to support blocklist mechanism, which is experimental currently, so we
* do not want to expose this option in the documentation.
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
index 03ca1dfb570d8..a08263e988c32 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClusterDescriptorTest.java
@@ -86,7 +86,8 @@ public FlinkKubeClient fromConfiguration(
server.createClient().inNamespace(NAMESPACE),
Executors.newSingleThreadScheduledExecutor());
}
- });
+ },
+ config -> {});
}
@Test
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DefaultKubernetesArtifactUploaderTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DefaultKubernetesArtifactUploaderTest.java
new file mode 100644
index 0000000000000..77bddf7937d7b
--- /dev/null
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DefaultKubernetesArtifactUploaderTest.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.artifact;
+
+import org.apache.flink.client.cli.ArtifactFetchOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.testutils.TestingUtils;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link DefaultKubernetesArtifactUploader}. */
+class DefaultKubernetesArtifactUploaderTest {
+
+ private final DefaultKubernetesArtifactUploader artifactUploader =
+ new DefaultKubernetesArtifactUploader();
+
+ @TempDir private Path tmpDir;
+
+ private Configuration config;
+
+ private DummyFs dummyFs;
+
+ @BeforeEach
+ void setup() throws IOException {
+ config = new Configuration();
+ config.set(KubernetesConfigOptions.LOCAL_UPLOAD_ENABLED, true);
+ config.set(KubernetesConfigOptions.LOCAL_UPLOAD_TARGET, getTargetDirUri());
+
+ dummyFs = (DummyFs) new org.apache.flink.core.fs.Path(getTargetDirUri()).getFileSystem();
+ dummyFs.resetCallCounters();
+ }
+
+ @Test
+ void testInvalidJobJar() {
+ String msg = "The 'pipeline.jars' config must contain one JAR.";
+
+ config.set(PipelineOptions.JARS, Collections.emptyList());
+ assertThatThrownBy(() -> artifactUploader.uploadAll(config))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(msg);
+
+ config.set(PipelineOptions.JARS, Arrays.asList("a", "b"));
+ assertThatThrownBy(() -> artifactUploader.uploadAll(config))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(msg);
+ }
+
+ @Test
+ void testUploadAllWithOneJobJar() throws Exception {
+ File jar = getFlinkKubernetesJar();
+ String localUri = "local://" + jar.getAbsolutePath();
+
+ config.set(PipelineOptions.JARS, Collections.singletonList(localUri));
+ artifactUploader.uploadAll(config);
+
+ assertJobJarUri(jar.getName());
+ }
+
+ @Test
+ void testUploadAllWithAdditionalArtifacts() throws Exception {
+ File jobJar = getFlinkKubernetesJar();
+ File addArtifact1 = TestingUtils.getClassFile(DefaultKubernetesArtifactUploader.class);
+ File addArtifact2 = TestingUtils.getClassFile(KubernetesUtils.class);
+ String localJobUri = "local://" + jobJar.getAbsolutePath();
+ String localAddArtUri = "local://" + addArtifact1.getAbsolutePath();
+ String nonLocalAddArtUri = "dummyfs://" + addArtifact2.getAbsolutePath();
+
+ config.set(PipelineOptions.JARS, Collections.singletonList(localJobUri));
+ config.set(
+ ArtifactFetchOptions.ARTIFACT_LIST,
+ Arrays.asList(nonLocalAddArtUri, localAddArtUri));
+ artifactUploader.uploadAll(config);
+
+ assertJobJarUri(jobJar.getName());
+
+ List additionalArtifactsResult = config.get(ArtifactFetchOptions.ARTIFACT_LIST);
+ assertThat(additionalArtifactsResult).hasSize(2);
+ assertThat(additionalArtifactsResult)
+ .containsExactlyInAnyOrder(
+ nonLocalAddArtUri, "dummyfs:" + tmpDir.resolve(addArtifact1.getName()));
+ }
+
+ @Test
+ void testMissingTargetConf() {
+ config.removeConfig(KubernetesConfigOptions.LOCAL_UPLOAD_TARGET);
+
+ assertThatThrownBy(() -> artifactUploader.upload(config, "local:///tmp/my-artifact.jar"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Setting 'kubernetes.artifacts.local-upload-target' to a valid remote path is required.");
+ }
+
+ @Test
+ void testRemoteUri() throws Exception {
+ config.removeConfig(KubernetesConfigOptions.LOCAL_UPLOAD_TARGET);
+ String remoteUri = "s3://my-bucket/my-artifact.jar";
+
+ String finalUri = artifactUploader.upload(config, remoteUri);
+
+ assertThat(finalUri).isEqualTo(remoteUri);
+ }
+
+ @Test
+ void testUpload() throws Exception {
+ File jar = getFlinkKubernetesJar();
+ String localUri = "local://" + jar.getAbsolutePath();
+
+ String expectedUri = "dummyfs:" + tmpDir.resolve(jar.getName());
+ String resultUri = artifactUploader.upload(config, localUri);
+
+ assertThat(resultUri).isEqualTo(expectedUri);
+ }
+
+ @Test
+ void testUploadNoOverwrite() throws Exception {
+ File jar = getFlinkKubernetesJar();
+ String localUri = "local://" + jar.getAbsolutePath();
+ Files.createFile(tmpDir.resolve(jar.getName()));
+
+ artifactUploader.upload(config, localUri);
+
+ assertThat(dummyFs.getExistsCallCounter()).isOne();
+ assertThat(dummyFs.getCreateCallCounter()).isZero();
+ }
+
+ @Test
+ void testUploadOverwrite() throws Exception {
+ File jar = getFlinkKubernetesJar();
+ String localUri = "local://" + jar.getAbsolutePath();
+ Files.createFile(tmpDir.resolve(jar.getName()));
+
+ config.set(KubernetesConfigOptions.LOCAL_UPLOAD_OVERWRITE, true);
+ artifactUploader.upload(config, localUri);
+
+ assertThat(dummyFs.getExistsCallCounter()).isEqualTo(2);
+ assertThat(dummyFs.getCreateCallCounter()).isOne();
+ }
+
+ @Test
+ void testUpdateConfig() {
+ List artifactList =
+ Arrays.asList("local:///tmp/artifact1.jar", "s3://my-bucket/artifact2.jar");
+ Configuration config = new Configuration();
+ config.set(ArtifactFetchOptions.ARTIFACT_LIST, artifactList);
+
+ List uploadedArtifactList = new ArrayList<>(artifactList);
+ uploadedArtifactList.set(0, getTargetDirUri() + "/artifact1.jar");
+ artifactUploader.updateConfig(
+ config, ArtifactFetchOptions.ARTIFACT_LIST, uploadedArtifactList);
+
+ assertThat(config.get(ArtifactFetchOptions.ARTIFACT_LIST)).isEqualTo(uploadedArtifactList);
+ }
+
+ @Test
+ void testNoUpdateConfig() {
+ List artifactList = Collections.singletonList("s3://my-bucket/my-artifact.jar");
+ Configuration config = new Configuration();
+ config.set(ArtifactFetchOptions.ARTIFACT_LIST, artifactList);
+
+ artifactUploader.updateConfig(config, ArtifactFetchOptions.ARTIFACT_LIST, artifactList);
+
+ assertThat(config.get(ArtifactFetchOptions.ARTIFACT_LIST)).isEqualTo(artifactList);
+ }
+
+ private String getTargetDirUri() {
+ return "dummyfs://" + tmpDir;
+ }
+
+ private File getFlinkKubernetesJar() throws IOException {
+ return TestingUtils.getFileFromTargetDir(
+ DefaultKubernetesArtifactUploader.class,
+ p ->
+ org.apache.flink.util.FileUtils.isJarFile(p)
+ && p.toFile().getName().startsWith("flink-kubernetes"));
+ }
+
+ private void assertJobJarUri(String filename) {
+ String expectedUri = "dummyfs:" + tmpDir.resolve(filename);
+
+ List result = config.get(PipelineOptions.JARS);
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0)).isEqualTo(expectedUri);
+ }
+}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DummyFs.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DummyFs.java
new file mode 100644
index 0000000000000..cd80607587874
--- /dev/null
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DummyFs.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.artifact;
+
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.fs.local.LocalFileSystem;
+
+import java.io.IOException;
+import java.net.URI;
+
+/** Dummy filesystem to test local artifact upload. */
+public class DummyFs extends LocalFileSystem {
+
+ static final URI FS_URI = URI.create("dummyfs:///");
+
+ private int existsCallCounter;
+
+ private int createCallCounter;
+
+ @Override
+ public URI getUri() {
+ return FS_URI;
+ }
+
+ @Override
+ public boolean exists(Path f) throws IOException {
+ ++existsCallCounter;
+ return super.exists(f);
+ }
+
+ @Override
+ public FSDataOutputStream create(Path filePath, WriteMode overwrite) throws IOException {
+ ++createCallCounter;
+ return super.create(filePath, overwrite);
+ }
+
+ public void resetCallCounters() {
+ createCallCounter = 0;
+ existsCallCounter = 0;
+ }
+
+ public int getExistsCallCounter() {
+ return existsCallCounter;
+ }
+
+ public int getCreateCallCounter() {
+ return createCallCounter;
+ }
+}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DummyFsFactory.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DummyFsFactory.java
new file mode 100644
index 0000000000000..f2af0dd3592f4
--- /dev/null
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/artifact/DummyFsFactory.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.artifact;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemFactory;
+
+import java.io.IOException;
+import java.net.URI;
+
+/** Dummy filesystem factory to test local artifact upload. */
+public class DummyFsFactory implements FileSystemFactory {
+
+ @Override
+ public String getScheme() {
+ return DummyFs.FS_URI.getScheme();
+ }
+
+ @Override
+ public FileSystem create(URI fsUri) throws IOException {
+ return new DummyFs();
+ }
+}
diff --git a/flink-kubernetes/src/test/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory b/flink-kubernetes/src/test/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory
new file mode 100644
index 0000000000000..d4a677f649efe
--- /dev/null
+++ b/flink-kubernetes/src/test/resources/META-INF/services/org.apache.flink.core.fs.FileSystemFactory
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.flink.kubernetes.artifact.DummyFsFactory