This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 8f53441a [FLINK-29288] Make it possible to use job jars in the system
classpath
8f53441a is described below
commit 8f53441a4978eeb38dc5ef229c179cc60598ce87
Author: Yaroslav Tkachenko <[email protected]>
AuthorDate: Thu Sep 22 08:43:38 2022 -0700
[FLINK-29288] Make it possible to use job jars in the system classpath
---
.dockerignore | 2 +
.gitignore | 2 +
docs/content/docs/custom-resource/reference.md | 2 +-
.../operator/config/FlinkConfigBuilder.java | 11 +++--
.../kubernetes/operator/crd/spec/JobSpec.java | 5 +-
.../operator/service/AbstractFlinkService.java | 34 +++++++++++++-
.../operator/validation/DefaultValidator.java | 4 --
.../src/main/resources/noop.jar | Bin 0 -> 305 bytes
.../operator/config/FlinkConfigBuilderTest.java | 13 ++++++
.../controller/FlinkDeploymentControllerTest.java | 2 +-
.../operator/validation/DefaultValidatorTest.java | 5 --
.../decorators/UserLibMountDecorator.java | 4 ++
.../StandaloneKubernetesJobManagerParameters.java | 5 ++
.../decorators/UserLibMountDecoratorTest.java | 51 +++++++++++++++++----
.../StandaloneKubernetesJobManagerFactoryTest.java | 3 ++
15 files changed, 116 insertions(+), 27 deletions(-)
diff --git a/.dockerignore b/.dockerignore
index 5013ecf0..60e24e7e 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -21,3 +21,5 @@
**/.idea
**/*.iml
**/*.DS_Store
+
+!flink-kubernetes-operator/src/main/resources/noop.jar
diff --git a/.gitignore b/.gitignore
index bb85f4bb..0d64c613 100644
--- a/.gitignore
+++ b/.gitignore
@@ -36,3 +36,5 @@ buildNumber.properties
.idea
*.iml
*.DS_Store
+
+!flink-kubernetes-operator/src/main/resources/noop.jar
diff --git a/docs/content/docs/custom-resource/reference.md
b/docs/content/docs/custom-resource/reference.md
index b2399505..400c078a 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -112,7 +112,7 @@ This page serves as a full reference for FlinkDeployment
custom resource definit
| Parameter | Type | Docs |
| ----------| ---- | ---- |
-| jarURI | java.lang.String | URI of the job jar within the Flink docker
container. For example: Example:
local:///opt/flink/examples/streaming/StateMachineExample.jar |
+| jarURI | java.lang.String | Optional URI of the job jar within the Flink
docker container. For example:
local:///opt/flink/examples/streaming/StateMachineExample.jar. If not specified
the job jar should be available in the system classpath. |
| parallelism | int | Parallelism of the Flink job. |
| entryClass | java.lang.String | Fully qualified main class name of the Flink
job. |
| args | java.lang.String[] | Arguments for the Flink job main class. |
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
index fff4b71d..d1423eb4 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java
@@ -255,8 +255,13 @@ public class FlinkConfigBuilder {
JobSpec jobSpec = spec.getJob();
effectiveConfig.set(
DeploymentOptions.TARGET,
KubernetesDeploymentTarget.APPLICATION.getName());
- final URI uri = new URI(jobSpec.getJarURI());
- effectiveConfig.set(PipelineOptions.JARS,
Collections.singletonList(uri.toString()));
+
+ if (jobSpec.getJarURI() != null) {
+ final URI uri = new URI(jobSpec.getJarURI());
+ effectiveConfig.set(
+ PipelineOptions.JARS,
Collections.singletonList(uri.toString()));
+ }
+
effectiveConfig.set(CoreOptions.DEFAULT_PARALLELISM,
getParallelism());
if (jobSpec.getAllowNonRestoredState() != null) {
@@ -280,7 +285,7 @@ public class FlinkConfigBuilder {
StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE,
spec.getJob() == null ? SESSION : APPLICATION);
- if (spec.getJob() != null) {
+ if (spec.getJob() != null && spec.getJob().getJarURI() != null) {
effectiveConfig.set(
PipelineOptions.CLASSPATHS,
Collections.singletonList(getStandaloneJarURI(spec.getJob())));
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java
index 3583a5e3..68aea7b4 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java
@@ -40,8 +40,9 @@ import lombok.NoArgsConstructor;
public class JobSpec implements Diffable<JobSpec> {
/**
- * URI of the job jar within the Flink docker container. For example:
Example:
- * local:///opt/flink/examples/streaming/StateMachineExample.jar
+ * Optional URI of the job jar within the Flink docker container. For
example:
+ * local:///opt/flink/examples/streaming/StateMachineExample.jar. If not
specified the job jar
+ * should be available in the system classpath.
*/
private String jarURI;
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index bd316290..2b8996af 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -96,10 +96,12 @@ import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
+import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.URI;
+import java.nio.file.Files;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -123,11 +125,13 @@ import static
org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLI
public abstract class AbstractFlinkService implements FlinkService {
private static final Logger LOG =
LoggerFactory.getLogger(AbstractFlinkService.class);
+ private static final String NOOP_JAR_FILENAME = "noop.jar";
protected final KubernetesClient kubernetesClient;
protected final FlinkConfigManager configManager;
private final ExecutorService executorService;
protected final ArtifactManager artifactManager;
+ private final String noopJarPath;
public AbstractFlinkService(
KubernetesClient kubernetesClient, FlinkConfigManager
configManager) {
@@ -137,6 +141,7 @@ public abstract class AbstractFlinkService implements
FlinkService {
this.executorService =
Executors.newFixedThreadPool(
4, new
ExecutorThreadFactory("Flink-RestClusterClient-IO"));
+ this.noopJarPath = copyNoopJar();
}
protected abstract PodList getJmPodList(String namespace, String
clusterId);
@@ -669,7 +674,7 @@ public abstract class AbstractFlinkService implements
FlinkService {
private JarUploadResponseBody uploadJar(
ObjectMeta objectMeta, FlinkSessionJobSpec spec, Configuration
conf) throws Exception {
String targetDir = artifactManager.generateJarDir(objectMeta, spec);
- File jarFile = artifactManager.fetch(spec.getJob().getJarURI(), conf,
targetDir);
+ File jarFile = artifactManager.fetch(findJarURI(spec.getJob()), conf,
targetDir);
Preconditions.checkArgument(
jarFile.exists(),
String.format("The jar file %s not exists",
jarFile.getAbsolutePath()));
@@ -706,6 +711,14 @@ public abstract class AbstractFlinkService implements
FlinkService {
}
}
+ private String findJarURI(JobSpec jobSpec) {
+ if (jobSpec.getJarURI() != null) {
+ return jobSpec.getJarURI();
+ } else {
+ return noopJarPath;
+ }
+ }
+
private void deleteJar(Configuration conf, String jarId) {
LOG.debug("Deleting the jar: {}", jarId);
try (RestClusterClient<String> clusterClient =
@@ -806,4 +819,23 @@ public abstract class AbstractFlinkService implements
FlinkService {
"RestoreFailed");
}
}
+
+ private String copyNoopJar() {
+ try {
+ InputStream noopJarSource =
+ AbstractFlinkService.class
+ .getClassLoader()
+ .getResourceAsStream(NOOP_JAR_FILENAME);
+
+ String noopJarDestination =
+ Files.createTempDirectory("flink").toString() + "/" +
NOOP_JAR_FILENAME;
+
+ LOG.debug("Copying noop jar to {}", noopJarDestination);
+ org.apache.commons.io.FileUtils.copyToFile(noopJarSource, new
File(noopJarDestination));
+
+ return noopJarDestination;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to copy noop jar", e);
+ }
+ }
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
index 496dbc1d..5486c7d6 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
@@ -180,10 +180,6 @@ public class DefaultValidator implements
FlinkResourceValidator {
return Optional.empty();
}
- if (StringUtils.isNullOrWhitespaceOnly(job.getJarURI())) {
- return Optional.of("Jar URI must be defined");
- }
-
Configuration configuration = Configuration.fromMap(confMap);
if (job.getUpgradeMode() == UpgradeMode.LAST_STATE
&& !FlinkUtils.isKubernetesHAActivated(configuration)) {
diff --git a/flink-kubernetes-operator/src/main/resources/noop.jar
b/flink-kubernetes-operator/src/main/resources/noop.jar
new file mode 100644
index 00000000..57bf8162
Binary files /dev/null and
b/flink-kubernetes-operator/src/main/resources/noop.jar differ
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
index a29018e3..69ade6fb 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java
@@ -327,6 +327,19 @@ public class FlinkConfigBuilderTest {
Assertions.assertEquals(12,
configuration.get(CoreOptions.DEFAULT_PARALLELISM));
}
+ @Test
+ public void testApplyJobOrSessionSpecWithNoJar() throws Exception {
+ FlinkDeployment deploymentClone =
ReconciliationUtils.clone(flinkDeployment);
+ deploymentClone.getSpec().getJob().setJarURI(null);
+
+ var configuration =
+ new FlinkConfigBuilder(deploymentClone, new Configuration())
+ .applyJobOrSessionSpec()
+ .build();
+
+ Assertions.assertNull(configuration.get(PipelineOptions.JARS));
+ }
+
@Test
public void testAllowNonRestoredStateInSpecOverrideInFlinkConf() throws
URISyntaxException {
flinkDeployment.getSpec().getJob().setAllowNonRestoredState(false);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
index f5bd8d4c..36003756 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -654,7 +654,7 @@ public class FlinkDeploymentControllerTest {
// Make sure we do validation before getting effective config in
reconcile().
appCluster.getSpec().getJobManager().setReplicas(1);
- appCluster.getSpec().getJob().setJarURI(null);
+ appCluster.getSpec().getJob().setParallelism(0);
// Verify the saved rest port in lastReconciledSpec is actually used
in observe() by
// utilizing listJobConsumer
appCluster.getSpec().getFlinkConfiguration().put(RestOptions.PORT.key(),
"12345");
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
index 4e47e0aa..0b2c3361 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
@@ -75,11 +75,6 @@ public class DefaultValidatorTest {
dep -> dep.getMetadata().setName("session-cluster-1.13"),
"The FlinkDeployment name: session-cluster-1.13 is invalid,
must consist of lower case alphanumeric characters or '-', start with an
alphabetic character, and end with an alphanumeric character (e.g. 'my-name',
or 'abc-123'), and the length must be no more than 45 characters.");
- // Test job validation
- testError(dep -> dep.getSpec().getJob().setJarURI(null), "Jar URI must
be defined");
- testError(dep -> dep.getSpec().getJob().setJarURI(""), "Jar URI must
be defined");
- testError(dep -> dep.getSpec().getJob().setJarURI(" "), "Jar URI must
be defined");
-
testError(
dep -> dep.getSpec().getJob().setState(JobState.SUSPENDED),
"Job must start in running state");
diff --git
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/UserLibMountDecorator.java
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/UserLibMountDecorator.java
index 207637ac..686bf0f1 100644
---
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/UserLibMountDecorator.java
+++
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/UserLibMountDecorator.java
@@ -58,6 +58,10 @@ public class UserLibMountDecorator extends
AbstractKubernetesStepDecorator {
return flinkPod;
}
+ if (!kubernetesJobManagerParameters.isPipelineClasspathDefined()) {
+ return flinkPod;
+ }
+
final Volume userLibVolume =
new VolumeBuilder()
.withName(USER_LIB_VOLUME)
diff --git
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesJobManagerParameters.java
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesJobManagerParameters.java
index e473d47f..a46172f4 100644
---
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesJobManagerParameters.java
+++
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesJobManagerParameters.java
@@ -20,6 +20,7 @@ package
org.apache.flink.kubernetes.operator.kubeclient.parameters;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import
org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters;
import
org.apache.flink.kubernetes.operator.standalone.StandaloneKubernetesConfigOptionsInternal;
@@ -87,4 +88,8 @@ public class StandaloneKubernetesJobManagerParameters extends
KubernetesJobManag
}
return null;
}
+
+ public boolean isPipelineClasspathDefined() {
+ return flinkConfig.contains(PipelineOptions.CLASSPATHS);
+ }
}
diff --git
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/UserLibMountDecoratorTest.java
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/UserLibMountDecoratorTest.java
index 1a867521..ba229965 100644
---
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/UserLibMountDecoratorTest.java
+++
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/decorators/UserLibMountDecoratorTest.java
@@ -19,6 +19,7 @@ package
org.apache.flink.kubernetes.operator.kubeclient.decorators;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.kubernetes.kubeclient.FlinkPod;
import
org.apache.flink.kubernetes.operator.kubeclient.parameters.StandaloneKubernetesJobManagerParameters;
import
org.apache.flink.kubernetes.operator.standalone.StandaloneKubernetesConfigOptionsInternal;
@@ -30,6 +31,9 @@ import io.fabric8.kubernetes.api.model.VolumeBuilder;
import io.fabric8.kubernetes.api.model.VolumeMount;
import org.junit.jupiter.api.Test;
+import java.util.Collections;
+import java.util.List;
+
import static org.junit.jupiter.api.Assertions.assertEquals;
/** @link UserLibMountDecorator unit tests */
@@ -38,8 +42,9 @@ public class UserLibMountDecoratorTest {
@Test
public void testVolumeAddedApplicationMode() {
StandaloneKubernetesJobManagerParameters jmParameters =
- createJmParamsWithClusterMode(
-
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION);
+ createJmParams(
+
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION,
+ Collections.singletonList("/path"));
UserLibMountDecorator decorator = new
UserLibMountDecorator(jmParameters);
@@ -60,8 +65,9 @@ public class UserLibMountDecoratorTest {
@Test
public void testVolumeNotAddedSessionMode() {
StandaloneKubernetesJobManagerParameters jmParameters =
- createJmParamsWithClusterMode(
-
StandaloneKubernetesConfigOptionsInternal.ClusterMode.SESSION);
+ createJmParams(
+
StandaloneKubernetesConfigOptionsInternal.ClusterMode.SESSION,
+ Collections.singletonList("/path"));
UserLibMountDecorator decorator = new
UserLibMountDecorator(jmParameters);
FlinkPod baseFlinkPod = new FlinkPod.Builder().build();
@@ -76,8 +82,9 @@ public class UserLibMountDecoratorTest {
@Test
public void testVolumeNotAddedExistingVolumeMount() {
StandaloneKubernetesJobManagerParameters jmParameters =
- createJmParamsWithClusterMode(
-
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION);
+ createJmParams(
+
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION,
+ Collections.singletonList("/path"));
UserLibMountDecorator decorator = new
UserLibMountDecorator(jmParameters);
final String volName = "flink-artifact";
@@ -115,11 +122,35 @@ public class UserLibMountDecoratorTest {
assertEquals(1,
decoratedPod.getPodWithoutMainContainer().getSpec().getVolumes().size());
}
- private StandaloneKubernetesJobManagerParameters
createJmParamsWithClusterMode(
- StandaloneKubernetesConfigOptionsInternal.ClusterMode clusterMode)
{
- return new StandaloneKubernetesJobManagerParameters(
+ @Test
+ public void testVolumeNotAddedNoPipelineClasspaths() {
+ StandaloneKubernetesJobManagerParameters jmParameters =
+ createJmParams(
+
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION, null);
+ UserLibMountDecorator decorator = new
UserLibMountDecorator(jmParameters);
+
+ FlinkPod baseFlinkPod = new FlinkPod.Builder().build();
+ assertEquals(0,
baseFlinkPod.getMainContainer().getVolumeMounts().size());
+ assertEquals(0,
baseFlinkPod.getPodWithoutMainContainer().getSpec().getVolumes().size());
+
+ FlinkPod decoratedPod = decorator.decorateFlinkPod(baseFlinkPod);
+ assertEquals(0,
decoratedPod.getMainContainer().getVolumeMounts().size());
+ assertEquals(0,
decoratedPod.getPodWithoutMainContainer().getSpec().getVolumes().size());
+ }
+
+ private StandaloneKubernetesJobManagerParameters createJmParams(
+ StandaloneKubernetesConfigOptionsInternal.ClusterMode clusterMode,
+ List<String> pipelineClasspaths) {
+ Configuration configuration =
new Configuration()
-
.set(StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE, clusterMode),
+
.set(StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE, clusterMode);
+
+ if (pipelineClasspaths != null && !pipelineClasspaths.isEmpty()) {
+ configuration.set(PipelineOptions.CLASSPATHS, pipelineClasspaths);
+ }
+
+ return new StandaloneKubernetesJobManagerParameters(
+ configuration,
new ClusterSpecification.ClusterSpecificationBuilder()
.createClusterSpecification());
}
diff --git
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/factory/StandaloneKubernetesJobManagerFactoryTest.java
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/factory/StandaloneKubernetesJobManagerFactoryTest.java
index 80b5ab80..de322891 100644
---
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/factory/StandaloneKubernetesJobManagerFactoryTest.java
+++
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/factory/StandaloneKubernetesJobManagerFactoryTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.kubernetes.operator.kubeclient.factory;
+import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.kubeclient.FlinkPod;
import
org.apache.flink.kubernetes.kubeclient.KubernetesJobManagerSpecification;
@@ -46,6 +47,7 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -67,6 +69,7 @@ public class StandaloneKubernetesJobManagerFactoryTest
extends ParametersTestBas
public void setup() throws Exception {
setupFlinkConfig();
flinkConfig.set(KubernetesConfigOptions.FLINK_CONF_DIR,
"/missing/dir");
+ flinkConfig.set(PipelineOptions.CLASSPATHS,
Collections.singletonList("/path"));
FlinkPod podTemplate = createPodTemplate();
StandaloneKubernetesJobManagerParameters tmParameters =
new StandaloneKubernetesJobManagerParameters(
