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 04f5762 [FLINK-26894] Support custom validator implementations
04f5762 is described below
commit 04f57627e3ea4a32c8682ce8d695319a72ebabc8
Author: Nicholas Jiang <[email protected]>
AuthorDate: Thu Apr 7 02:33:05 2022 +0800
[FLINK-26894] Support custom validator implementations
---
flink-kubernetes-operator/pom.xml | 20 ++++++
.../flink/kubernetes/operator/FlinkOperator.java | 11 +--
.../operator/config/OperatorConfigOptions.java | 4 +-
.../controller/FlinkDeploymentController.java | 18 +++--
.../controller/FlinkSessionJobController.java | 27 +++++---
.../kubernetes/operator/utils/ValidatorUtils.java | 59 ++++++++++++++++
.../validation/FlinkResourceValidator.java | 5 +-
.../src/test/assembly/test-validator-assembly.xml | 40 +++++++++++
.../flink/kubernetes/operator/TestUtils.java | 31 +++++++++
.../controller/FlinkDeploymentControllerTest.java | 4 +-
.../operator/utils/ValidatorUtilsTest.java | 79 ++++++++++++++++++++++
.../operator/validation/TestValidator.java} | 29 +++-----
...etes.operator.validation.FlinkResourceValidator | 16 +++++
13 files changed, 302 insertions(+), 41 deletions(-)
diff --git a/flink-kubernetes-operator/pom.xml
b/flink-kubernetes-operator/pom.xml
index 1ca0e66..d0181e7 100644
--- a/flink-kubernetes-operator/pom.xml
+++ b/flink-kubernetes-operator/pom.xml
@@ -262,6 +262,26 @@ under the License.
</artifactItems>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>create-test-validator-jar</id>
+ <phase>process-test-classes</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <configuration>
+ <finalName>test-validator</finalName>
+ <attach>false</attach>
+ <descriptors>
+
<descriptor>src/test/assembly/test-validator-assembly.xml</descriptor>
+ </descriptors>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
<profiles>
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index a569733..2591cac 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -34,7 +34,7 @@ import
org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFact
import
org.apache.flink.kubernetes.operator.reconciler.sessionjob.FlinkSessionJobReconciler;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
-import org.apache.flink.kubernetes.operator.validation.DefaultValidator;
+import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
import org.apache.flink.runtime.util.EnvironmentInformation;
@@ -47,6 +47,7 @@ import
io.javaoperatorsdk.operator.config.runtime.DefaultConfigurationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Set;
import java.util.concurrent.Executors;
/** Main Class for Flink native k8s operator. */
@@ -60,6 +61,7 @@ public class FlinkOperator {
private final FlinkService flinkService;
private final ConfigurationService configurationService;
private final DefaultConfig defaultConfig;
+ private final Set<FlinkResourceValidator> validators;
public FlinkOperator() {
this(FlinkUtils.loadDefaultConfig());
@@ -75,10 +77,10 @@ public class FlinkOperator {
this.configurationService =
getConfigurationService(operatorConfiguration);
this.operator = new Operator(client, configurationService);
this.flinkService = new FlinkService(client, operatorConfiguration);
+ this.validators =
ValidatorUtils.discoverValidators(defaultConfig.getFlinkConfig());
}
private void registerDeploymentController() {
- FlinkResourceValidator validator = new DefaultValidator();
ReconcilerFactory reconcilerFactory =
new ReconcilerFactory(client, flinkService,
operatorConfiguration);
ObserverFactory observerFactory =
@@ -90,7 +92,7 @@ public class FlinkOperator {
defaultConfig,
operatorConfiguration,
client,
- validator,
+ validators,
reconcilerFactory,
observerFactory);
@@ -103,7 +105,6 @@ public class FlinkOperator {
}
private void registerSessionJobController() {
- FlinkResourceValidator validator = new DefaultValidator();
Reconciler<FlinkSessionJob> reconciler =
new FlinkSessionJobReconciler(client, flinkService,
operatorConfiguration);
Observer<FlinkSessionJob> observer = new SessionJobObserver();
@@ -112,7 +113,7 @@ public class FlinkOperator {
defaultConfig,
operatorConfiguration,
client,
- validator,
+ validators,
reconciler,
observer);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
index 40058bd..65d7cd3 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
@@ -33,7 +33,7 @@ public class OperatorConfigOptions {
.durationType()
.defaultValue(Duration.ofSeconds(60))
.withDescription(
- "The interval for the controller to reschedule the
reconcile process");
+ "The interval for the controller to reschedule the
reconcile process.");
public static final ConfigOption<Duration>
OPERATOR_OBSERVER_REST_READY_DELAY =
ConfigOptions.key("operator.observer.rest-ready.delay")
@@ -61,7 +61,7 @@ public class OperatorConfigOptions {
.durationType()
.defaultValue(Duration.ofSeconds(10))
.withDescription(
- "The interval before a savepoint trigger attempt
is marked as unsuccessful");
+ "The interval before a savepoint trigger attempt
is marked as unsuccessful.");
public static final ConfigOption<Duration>
OPERATOR_OBSERVER_FLINK_CLIENT_TIMEOUT =
ConfigOptions.key("operator.observer.flink.client.timeout")
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
index 8ebeb50..e470331 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
@@ -62,7 +62,7 @@ public class FlinkDeploymentController
private final KubernetesClient kubernetesClient;
- private final FlinkResourceValidator validator;
+ private final Set<FlinkResourceValidator> validators;
private final ReconcilerFactory reconcilerFactory;
private final ObserverFactory observerFactory;
private final DefaultConfig defaultConfig;
@@ -74,13 +74,13 @@ public class FlinkDeploymentController
DefaultConfig defaultConfig,
FlinkOperatorConfiguration operatorConfiguration,
KubernetesClient kubernetesClient,
- FlinkResourceValidator validator,
+ Set<FlinkResourceValidator> validators,
ReconcilerFactory reconcilerFactory,
ObserverFactory observerFactory) {
this.defaultConfig = defaultConfig;
this.operatorConfiguration = operatorConfiguration;
this.kubernetesClient = kubernetesClient;
- this.validator = validator;
+ this.validators = validators;
this.reconcilerFactory = reconcilerFactory;
this.observerFactory = observerFactory;
}
@@ -104,7 +104,7 @@ public class FlinkDeploymentController
FlinkDeployment originalCopy = ReconciliationUtils.clone(flinkApp);
try {
observerFactory.getOrCreate(flinkApp).observe(flinkApp, context);
- Optional<String> validationError =
validator.validateDeployment(flinkApp);
+ Optional<String> validationError = validateDeployment(flinkApp);
if (validationError.isPresent()) {
LOG.error("Validation failed: " + validationError.get());
ReconciliationUtils.updateForReconciliationError(flinkApp,
validationError.get());
@@ -169,4 +169,14 @@ public class FlinkDeploymentController
public void setControllerConfig(FlinkControllerConfig<FlinkDeployment>
config) {
this.controllerConfig = config;
}
+
+ private Optional<String> validateDeployment(FlinkDeployment deployment) {
+ Optional<String> validationError = Optional.empty();
+ for (FlinkResourceValidator validator : validators) {
+ if ((validationError =
validator.validateDeployment(deployment)).isPresent()) {
+ break;
+ }
+ }
+ return validationError;
+ }
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
index 1e563fc..b905b13 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
@@ -68,7 +68,7 @@ public class FlinkSessionJobController
private final KubernetesClient kubernetesClient;
- private final FlinkResourceValidator validator;
+ private final Set<FlinkResourceValidator> validators;
private final Reconciler<FlinkSessionJob> reconciler;
private final Observer<FlinkSessionJob> observer;
private final DefaultConfig defaultConfig;
@@ -80,13 +80,13 @@ public class FlinkSessionJobController
DefaultConfig defaultConfig,
FlinkOperatorConfiguration operatorConfiguration,
KubernetesClient kubernetesClient,
- FlinkResourceValidator validator,
+ Set<FlinkResourceValidator> validators,
Reconciler<FlinkSessionJob> reconciler,
Observer<FlinkSessionJob> observer) {
this.defaultConfig = defaultConfig;
this.operatorConfiguration = operatorConfiguration;
this.kubernetesClient = kubernetesClient;
- this.validator = validator;
+ this.validators = validators;
this.reconciler = reconciler;
this.observer = observer;
}
@@ -102,11 +102,7 @@ public class FlinkSessionJobController
LOG.info("Starting reconciliation");
FlinkSessionJob originalCopy =
ReconciliationUtils.clone(flinkSessionJob);
observer.observe(flinkSessionJob, context);
- Optional<String> validationError =
- validator.validateSessionJob(
- flinkSessionJob,
- OperatorUtils.getSecondaryResource(
- flinkSessionJob, context,
operatorConfiguration));
+ Optional<String> validationError = validateSessionJob(flinkSessionJob,
context);
if (validationError.isPresent()) {
LOG.error("Validation failed: " + validationError.get());
ReconciliationUtils.updateForReconciliationError(
@@ -242,4 +238,19 @@ public class FlinkSessionJobController
private Map<String, Function<FlinkSessionJob, List<String>>>
clusterToSessionJobIndexer() {
return Map.of(CLUSTER_ID_INDEX, sessionJob ->
List.of(sessionJob.getSpec().getClusterId()));
}
+
+ private Optional<String> validateSessionJob(FlinkSessionJob sessionJob,
Context context) {
+ Optional<String> validationError = Optional.empty();
+ for (FlinkResourceValidator validator : validators) {
+ if ((validationError =
+ validator.validateSessionJob(
+ sessionJob,
+ OperatorUtils.getSecondaryResource(
+ sessionJob, context,
operatorConfiguration)))
+ .isPresent()) {
+ break;
+ }
+ }
+ return validationError;
+ }
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/ValidatorUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/ValidatorUtils.java
new file mode 100644
index 0000000..fa5077f
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/ValidatorUtils.java
@@ -0,0 +1,59 @@
+/*
+ * 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.operator.utils;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.plugin.PluginUtils;
+import org.apache.flink.kubernetes.operator.validation.DefaultValidator;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Validator utilities. */
+public final class ValidatorUtils {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(FlinkUtils.class);
+
+ public static Set<FlinkResourceValidator> discoverValidators(Configuration
configuration) {
+ Set<FlinkResourceValidator> resourceValidators = new HashSet<>();
+ resourceValidators.add(new DefaultValidator());
+ DefaultValidator defaultValidator = new DefaultValidator();
+ defaultValidator.configure(configuration);
+ resourceValidators.add(defaultValidator);
+ PluginUtils.createPluginManagerFromRootFolder(configuration)
+ .load(FlinkResourceValidator.class)
+ .forEachRemaining(
+ validator -> {
+ LOG.info(
+ "Discovered resource validator from plugin
directory[{}]: {}.",
+ System.getenv()
+ .getOrDefault(
+
ConfigConstants.ENV_FLINK_PLUGINS_DIR,
+
ConfigConstants.DEFAULT_FLINK_PLUGINS_DIRS),
+ validator.getClass().getName());
+ validator.configure(configuration);
+ resourceValidators.add(validator);
+ });
+ return resourceValidators;
+ }
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/FlinkResourceValidator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/FlinkResourceValidator.java
index c5c13c1..6dacb25 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/FlinkResourceValidator.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/FlinkResourceValidator.java
@@ -17,18 +17,19 @@
package org.apache.flink.kubernetes.operator.validation;
+import org.apache.flink.core.plugin.Plugin;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import java.util.Optional;
/** Validator for different resources. */
-public interface FlinkResourceValidator {
+public interface FlinkResourceValidator extends Plugin {
/**
* Validate and return optional error.
*
- * @param deployment
+ * @param deployment A Flink application or session cluster deployment.
* @return Optional error string, should be present iff validation
resulted in an error
*/
Optional<String> validateDeployment(FlinkDeployment deployment);
diff --git
a/flink-kubernetes-operator/src/test/assembly/test-validator-assembly.xml
b/flink-kubernetes-operator/src/test/assembly/test-validator-assembly.xml
new file mode 100644
index 0000000..b4abc44
--- /dev/null
+++ b/flink-kubernetes-operator/src/test/assembly/test-validator-assembly.xml
@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+
+<assembly>
+ <id>test-jar</id>
+ <formats>
+ <format>jar</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>${project.build.testOutputDirectory}</directory>
+ <outputDirectory>/</outputDirectory>
+ <!-- the service impl -->
+ <includes>
+
<include>org/apache/flink/kubernetes/operator/validation/TestValidator.class</include>
+ </includes>
+ </fileSet>
+ <fileSet>
+ <!-- declaring the service impl -->
+
<directory>${project.basedir}/src/test/resources/META-INF/services</directory>
+ <outputDirectory>/META-INF/services</outputDirectory>
+ </fileSet>
+ </fileSets>
+</assembly>
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
index 929afb1..b0bc19e 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
@@ -52,6 +52,7 @@ import io.fabric8.kubernetes.api.model.apps.DeploymentStatus;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import java.lang.reflect.Field;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -291,4 +292,34 @@ public class TestUtils {
}
};
}
+
+ // This code is taken slightly modified from:
http://stackoverflow.com/a/7201825/568695
+ // it changes the environment variables of this JVM. Use only for testing
purposes!
+ @SuppressWarnings("unchecked")
+ public static void setEnv(Map<String, String> newEnv) {
+ try {
+ Map<String, String> env = System.getenv();
+ Class<?> clazz = env.getClass();
+ Field field = clazz.getDeclaredField("m");
+ field.setAccessible(true);
+ Map<String, String> map = (Map<String, String>) field.get(env);
+ map.clear();
+ map.putAll(newEnv);
+ // only for Windows
+ Class<?> processEnvironmentClass =
Class.forName("java.lang.ProcessEnvironment");
+ try {
+ Field theCaseInsensitiveEnvironmentField =
+
processEnvironmentClass.getDeclaredField("theCaseInsensitiveEnvironment");
+ theCaseInsensitiveEnvironmentField.setAccessible(true);
+ Map<String, String> ciEnv =
+ (Map<String, String>)
theCaseInsensitiveEnvironmentField.get(null);
+ ciEnv.clear();
+ ciEnv.putAll(newEnv);
+ } catch (NoSuchFieldException ignored) {
+ }
+
+ } catch (Exception e1) {
+ throw new RuntimeException(e1);
+ }
+ }
}
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 3b358ef..351c510 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
@@ -37,7 +37,7 @@ import
org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import
org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
-import org.apache.flink.kubernetes.operator.validation.DefaultValidator;
+import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import io.fabric8.kubernetes.api.model.EventBuilder;
@@ -607,7 +607,7 @@ public class FlinkDeploymentControllerTest {
defaultConfig,
operatorConfiguration,
kubernetesClient,
- new DefaultValidator(),
+
ValidatorUtils.discoverValidators(defaultConfig.getFlinkConfig()),
new ReconcilerFactory(
kubernetesClient, flinkService,
operatorConfiguration),
new ObserverFactory(
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ValidatorUtilsTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ValidatorUtilsTest.java
new file mode 100644
index 0000000..9e00448
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ValidatorUtilsTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.operator.utils;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.kubernetes.operator.validation.DefaultValidator;
+import org.apache.flink.kubernetes.operator.validation.TestValidator;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/** Test class for {@link ValidatorUtils}. */
+public class ValidatorUtilsTest {
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private static final String VALIDATOR_NAME = "test-validator";
+ private static final String VALIDATOR_JAR = VALIDATOR_NAME +
"-test-jar.jar";
+
+ @Test
+ public void testDiscoverValidators() throws IOException {
+ File validatorRootFolder = temporaryFolder.newFolder();
+ File testValidatorFolder = new File(validatorRootFolder,
VALIDATOR_NAME);
+ assertTrue(testValidatorFolder.mkdirs());
+ File testValidatorJar = new File("target", VALIDATOR_JAR);
+ assertTrue(testValidatorJar.exists());
+ Files.copy(
+ testValidatorJar.toPath(),
+ Paths.get(testValidatorFolder.toString(), VALIDATOR_JAR));
+ Map<String, String> originalEnv = System.getenv();
+ try {
+ Map<String, String> systemEnv = new HashMap<>(originalEnv);
+ systemEnv.put(ConfigConstants.ENV_FLINK_PLUGINS_DIR,
validatorRootFolder.getPath());
+ TestUtils.setEnv(systemEnv);
+ assertEquals(
+ new HashSet<>(
+ Arrays.asList(
+ DefaultValidator.class.getName(),
+ TestValidator.class.getName())),
+ ValidatorUtils.discoverValidators(new
Configuration()).stream()
+ .map(v -> v.getClass().getName())
+ .collect(Collectors.toSet()));
+ } finally {
+ TestUtils.setEnv(originalEnv);
+ }
+ }
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/FlinkResourceValidator.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/TestValidator.java
similarity index 61%
copy from
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/FlinkResourceValidator.java
copy to
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/TestValidator.java
index c5c13c1..3c6515f 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/FlinkResourceValidator.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/TestValidator.java
@@ -22,24 +22,17 @@ import
org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import java.util.Optional;
-/** Validator for different resources. */
-public interface FlinkResourceValidator {
+/** Test validator implementation of {@link FlinkResourceValidator}. */
+public class TestValidator implements FlinkResourceValidator {
- /**
- * Validate and return optional error.
- *
- * @param deployment
- * @return Optional error string, should be present iff validation
resulted in an error
- */
- Optional<String> validateDeployment(FlinkDeployment deployment);
+ @Override
+ public Optional<String> validateDeployment(FlinkDeployment deployment) {
+ return Optional.empty();
+ }
- /**
- * Validate and return optional error.
- *
- * @param sessionJob the session job to be validated.
- * @param session the target session cluster of the session job to be
validated.
- * @return Optional error string, should be present iff validation
resulted in an error
- */
- Optional<String> validateSessionJob(
- FlinkSessionJob sessionJob, Optional<FlinkDeployment> session);
+ @Override
+ public Optional<String> validateSessionJob(
+ FlinkSessionJob sessionJob, Optional<FlinkDeployment> session) {
+ return Optional.empty();
+ }
}
diff --git
a/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator
b/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator
new file mode 100644
index 0000000..db688ac
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator
@@ -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.operator.validation.TestValidator
