You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/08/03 06:12:37 UTC
[flink-kubernetes-operator] branch main updated: [hotfix] Clean up warnings and unnecessary code in tests
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 e3def9d2 [hotfix] Clean up warnings and unnecessary code in tests
e3def9d2 is described below
commit e3def9d26c15d9755324b8d7dacd9d2d667ef96b
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Wed Aug 3 08:12:32 2022 +0200
[hotfix] Clean up warnings and unnecessary code in tests
---
.../kubernetes/operator/FlinkOperatorITCase.java | 17 ++-
.../flink/kubernetes/operator/TestUtils.java | 152 +++++++--------------
.../kubernetes/operator/TestingClusterClient.java | 2 -
.../kubernetes/operator/TestingFlinkService.java | 39 ++----
.../operator/TestingFlinkServiceFactory.java | 11 --
.../operator/config/FlinkConfigBuilderTest.java | 4 +-
.../operator/config/FlinkConfigManagerTest.java | 4 +-
.../controller/DeploymentRecoveryTest.java | 2 +-
.../controller/FlinkDeploymentControllerTest.java | 42 +++---
.../operator/controller/RollbackTest.java | 4 +-
.../TestingFlinkDeploymentController.java | 8 +-
.../metrics/KubernetesOperatorMetricGroupTest.java | 4 +-
.../operator/metrics/OperatorJosdkMetricsTest.java | 5 +-
.../operator/observer/SavepointObserverTest.java | 9 +-
.../deployment/ApplicationObserverTest.java | 13 +-
.../observer/deployment/SessionObserverTest.java | 3 +-
.../sessionjob/SessionJobObserverTest.java | 5 +-
.../deployment/ApplicationReconcilerTest.java | 31 ++---
.../sessionjob/SessionJobReconcilerTest.java | 22 ++-
.../service/StandaloneFlinkServiceTest.java | 20 +--
.../kubernetes/operator/utils/FlinkUtilsTest.java | 11 +-
.../operator/utils/ValidatorUtilsTest.java | 2 +-
.../operator/validation/DefaultValidatorTest.java | 58 ++++----
23 files changed, 177 insertions(+), 291 deletions(-)
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorITCase.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorITCase.java
index 038fe2aa..4902ca78 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorITCase.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorITCase.java
@@ -87,15 +87,14 @@ public class FlinkOperatorITCase {
await().atMost(1, MINUTES)
.untilAsserted(
- () -> {
- assertThat(
- client.apps()
- .deployments()
- .inNamespace(TEST_NAMESPACE)
- .withName(flinkDeployment.getMetadata().getName())
- .isReady(),
- is(true));
- });
+ () ->
+ assertThat(
+ client.apps()
+ .deployments()
+ .inNamespace(TEST_NAMESPACE)
+ .withName(flinkDeployment.getMetadata().getName())
+ .isReady(),
+ is(true)));
}
private static FlinkDeployment buildSessionCluster() {
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 9c9687b3..b79a57f0 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
@@ -44,6 +44,7 @@ import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.ContainerStatus;
import io.fabric8.kubernetes.api.model.ContainerStatusBuilder;
+import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
import io.fabric8.kubernetes.api.model.Pod;
@@ -84,7 +85,7 @@ import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
-import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/** Testing utilities. */
public class TestUtils {
@@ -243,128 +244,64 @@ public class TestUtils {
return deployment;
}
- public static Context createContextWithDeployment(@Nullable Deployment deployment) {
- return new Context() {
+ public static <T extends HasMetadata> Context<T> createContextWithDeployment(
+ @Nullable Deployment deployment) {
+ return new TestingContext<>() {
@Override
- public Optional<RetryInfo> getRetryInfo() {
- return Optional.empty();
- }
-
- @Override
- public Optional getSecondaryResource(Class expectedType, String eventSourceName) {
- return Optional.ofNullable(deployment);
- }
-
- @Override
- public Set getSecondaryResources(Class expectedType) {
- return null;
- }
-
- @Override
- public ControllerConfiguration getControllerConfiguration() {
- return null;
- }
-
- @Override
- public ManagedDependentResourceContext managedDependentResourceContext() {
- return null;
+ public Optional<T> getSecondaryResource(Class expectedType, String eventSourceName) {
+ return (Optional<T>) Optional.ofNullable(deployment);
}
};
}
- public static Context createEmptyContext() {
+ public static <T extends HasMetadata> Context<T> createEmptyContext() {
return createContextWithDeployment(null);
}
- public static Context createContextWithReadyJobManagerDeployment() {
+ public static <T extends HasMetadata> Context<T> createContextWithReadyJobManagerDeployment() {
return createContextWithDeployment(createDeployment(true));
}
- public static Context createContextWithInProgressDeployment() {
+ public static <T extends HasMetadata> Context<T> createContextWithInProgressDeployment() {
return createContextWithDeployment(createDeployment(false));
}
- public static Context createContextWithReadyFlinkDeployment() {
+ public static <T extends HasMetadata> Context<T> createContextWithReadyFlinkDeployment() {
return createContextWithReadyFlinkDeployment(new HashMap<>());
}
- public static Context createContextWithReadyFlinkDeployment(
+ public static <T extends HasMetadata> Context<T> createContextWithReadyFlinkDeployment(
Map<String, String> flinkDepConfig) {
- return new Context() {
+ return new TestingContext<>() {
@Override
- public Optional<RetryInfo> getRetryInfo() {
- return Optional.empty();
- }
-
- @Override
- public Optional getSecondaryResource(Class expectedType, String eventSourceName) {
+ public Optional<T> getSecondaryResource(Class expectedType, String eventSourceName) {
var session = buildSessionCluster();
session.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
session.getSpec().getFlinkConfiguration().putAll(flinkDepConfig);
session.getStatus()
.getReconciliationStatus()
.serializeAndSetLastReconciledSpec(session.getSpec(), session);
- return Optional.of(session);
- }
-
- @Override
- public Set getSecondaryResources(Class expectedType) {
- return null;
- }
-
- @Override
- public ControllerConfiguration getControllerConfiguration() {
- return null;
- }
-
- @Override
- public ManagedDependentResourceContext managedDependentResourceContext() {
- return null;
+ return (Optional<T>) Optional.of(session);
}
};
}
- public static Context createContextWithNotReadyFlinkDeployment() {
- return new Context() {
+ public static <T extends HasMetadata> Context<T> createContextWithNotReadyFlinkDeployment() {
+ return new TestingContext<>() {
@Override
- public Optional<RetryInfo> getRetryInfo() {
- return Optional.empty();
- }
-
- @Override
- public Optional getSecondaryResource(Class expectedType, String eventSourceName) {
+ public Optional<T> getSecondaryResource(Class expectedType, String eventSourceName) {
var session = buildSessionCluster();
session.getStatus()
.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.MISSING);
- return Optional.of(session);
- }
-
- @Override
- public Set getSecondaryResources(Class expectedType) {
- return null;
- }
-
- @Override
- public ControllerConfiguration getControllerConfiguration() {
- return null;
- }
-
- @Override
- public ManagedDependentResourceContext managedDependentResourceContext() {
- return null;
+ return (Optional<T>) Optional.of(session);
}
};
}
public static final String DEPLOYMENT_ERROR = "test deployment error message";
- public static Context createContextWithFailedJobManagerDeployment() {
- return new Context() {
- @Override
- public Optional<RetryInfo> getRetryInfo() {
- return Optional.empty();
- }
-
+ public static <T extends HasMetadata> Context<T> createContextWithFailedJobManagerDeployment() {
+ return new TestingContext<>() {
@Override
public Optional getSecondaryResource(Class expectedType, String eventSourceName) {
DeploymentStatus status = new DeploymentStatus();
@@ -387,21 +324,6 @@ public class TestUtils {
deployment.setStatus(status);
return Optional.of(deployment);
}
-
- @Override
- public Set getSecondaryResources(Class expectedType) {
- return null;
- }
-
- @Override
- public ControllerConfiguration getControllerConfiguration() {
- return null;
- }
-
- @Override
- public ManagedDependentResourceContext managedDependentResourceContext() {
- return null;
- }
};
}
@@ -501,4 +423,36 @@ public class TestUtils {
return returnValue;
}
}
+
+ /**
+ * Base context implementation for tests.
+ *
+ * @param <T> Resource type
+ */
+ public static class TestingContext<T extends HasMetadata> implements Context<T> {
+ @Override
+ public Optional<RetryInfo> getRetryInfo() {
+ return Optional.empty();
+ }
+
+ @Override
+ public <T1> Set<T1> getSecondaryResources(Class<T1> aClass) {
+ return null;
+ }
+
+ @Override
+ public <T1> Optional<T1> getSecondaryResource(Class<T1> aClass, String s) {
+ return Optional.empty();
+ }
+
+ @Override
+ public ControllerConfiguration<T> getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedDependentResourceContext managedDependentResourceContext() {
+ return null;
+ }
+ }
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingClusterClient.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingClusterClient.java
index 11754611..d96427e4 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingClusterClient.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingClusterClient.java
@@ -76,12 +76,10 @@ public class TestingClusterClient<T> extends RestClusterClient<T> {
CompletableFuture.completedFuture(
new JobResult.Builder().jobId(jobID).netRuntime(1).build());
- private final Configuration configuration;
private final T clusterId;
public TestingClusterClient(Configuration configuration, T clusterId) throws Exception {
super(configuration, clusterId, (c, e) -> new StandaloneClientHAServices("localhost"));
- this.configuration = configuration;
this.clusterId = clusterId;
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index 4ff0a33f..ba929289 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -53,13 +53,11 @@ import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
import org.apache.flink.util.SerializedThrowable;
+import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.api.model.PodList;
import io.fabric8.kubernetes.client.KubernetesClient;
-import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
-import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
import javax.annotation.Nullable;
@@ -109,34 +107,15 @@ public class TestingFlinkService extends AbstractFlinkService {
super(kubernetesClient, new FlinkConfigManager(new Configuration()));
}
- public Context getContext() {
- return new Context() {
- @Override
- public Optional<RetryInfo> getRetryInfo() {
- return Optional.empty();
- }
+ public <T extends HasMetadata> Context<T> getContext() {
+ return new TestUtils.TestingContext<>() {
@Override
- public Optional getSecondaryResource(Class aClass, String s) {
+ public Optional<T> getSecondaryResource(Class aClass, String s) {
if (jobs.isEmpty() && sessions.isEmpty()) {
return Optional.empty();
}
- return Optional.of(TestUtils.createDeployment(true));
- }
-
- @Override
- public Set getSecondaryResources(Class expectedType) {
- return null;
- }
-
- @Override
- public ControllerConfiguration getControllerConfiguration() {
- return null;
- }
-
- @Override
- public ManagedDependentResourceContext managedDependentResourceContext() {
- return null;
+ return (Optional<T>) Optional.of(TestUtils.createDeployment(true));
}
};
}
@@ -375,7 +354,7 @@ public class TestingFlinkService extends AbstractFlinkService {
Optional<Tuple2<String, JobStatusMessage>> jobOpt =
jobs.stream().filter(js -> js.f1.getJobId().equals(jobID)).findAny();
- if (!jobOpt.isPresent()) {
+ if (jobOpt.isEmpty()) {
throw new Exception("Job not found");
}
@@ -423,7 +402,7 @@ public class TestingFlinkService extends AbstractFlinkService {
Optional<Tuple2<String, JobStatusMessage>> jobOpt =
jobs.stream().filter(js -> js.f1.getJobId().equals(jobId)).findAny();
- if (!jobOpt.isPresent()) {
+ if (jobOpt.isEmpty()) {
throw new Exception("Job not found");
}
@@ -464,7 +443,7 @@ public class TestingFlinkService extends AbstractFlinkService {
public void markApplicationJobFailedWithError(JobID jobID, String error) throws Exception {
var job = jobs.stream().filter(tuple -> tuple.f1.getJobId().equals(jobID)).findFirst();
- if (!job.isPresent()) {
+ if (job.isEmpty()) {
throw new Exception("The target job missed");
}
var oldStatus = job.get().f1;
@@ -494,7 +473,7 @@ public class TestingFlinkService extends AbstractFlinkService {
}
@Override
- public Map<String, String> getClusterInfo(Configuration conf) throws Exception {
+ public Map<String, String> getClusterInfo(Configuration conf) {
return CLUSTER_INFO;
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkServiceFactory.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkServiceFactory.java
index e4e4d5e0..0f6574b1 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkServiceFactory.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkServiceFactory.java
@@ -22,17 +22,10 @@ import org.apache.flink.kubernetes.operator.crd.spec.KubernetesDeploymentMode;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.service.FlinkServiceFactory;
-import java.lang.module.Configuration;
-
/** Flink service factory mock for tests. */
public class TestingFlinkServiceFactory extends FlinkServiceFactory {
private final FlinkService flinkService;
- public TestingFlinkServiceFactory() {
- super(null, null);
- flinkService = new TestingFlinkService();
- }
-
public TestingFlinkServiceFactory(FlinkService flinkService) {
super(null, null);
this.flinkService = flinkService;
@@ -45,8 +38,4 @@ public class TestingFlinkServiceFactory extends FlinkServiceFactory {
public FlinkService getOrCreate(FlinkDeployment deployment) {
return flinkService;
}
-
- public FlinkService getOrCreate(Configuration configuration) {
- return flinkService;
- }
}
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 f33a9986..81423097 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
@@ -57,7 +57,6 @@ import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.file.Files;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -82,8 +81,7 @@ public class FlinkConfigBuilderTest {
final Container container0 = new Container();
container0.setName("container0");
final Pod pod0 =
- TestUtils.getTestPod(
- "pod0 hostname", "pod0 api version", Arrays.asList(container0));
+ TestUtils.getTestPod("pod0 hostname", "pod0 api version", List.of(container0));
flinkDeployment.getSpec().setPodTemplate(pod0);
flinkDeployment.getSpec().setIngress(IngressSpec.builder().template("test.com").build());
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
index 879dfccf..abeee8b3 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
@@ -26,7 +26,6 @@ import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
-import org.apache.flink.kubernetes.operator.crd.status.ReconciliationStatus;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.utils.Constants;
@@ -65,8 +64,7 @@ public class FlinkConfigManagerTest {
.key(),
"false")));
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
- ReconciliationStatus reconciliationStatus =
- deployment.getStatus().getReconciliationStatus();
+ var reconciliationStatus = deployment.getStatus().getReconciliationStatus();
deployment.getSpec().getFlinkConfiguration().put(testConf.key(), "reconciled");
reconciliationStatus.serializeAndSetLastReconciledSpec(deployment.getSpec(), deployment);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/DeploymentRecoveryTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/DeploymentRecoveryTest.java
index 7ecd28f4..5173fcfc 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/DeploymentRecoveryTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/DeploymentRecoveryTest.java
@@ -50,7 +50,7 @@ public class DeploymentRecoveryTest {
private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
private TestingFlinkService flinkService;
- private Context context;
+ private Context<FlinkDeployment> context;
private TestingFlinkDeploymentController testController;
private KubernetesClient kubernetesClient;
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 866ecbd5..b6aeb02c 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
@@ -79,7 +79,7 @@ public class FlinkDeploymentControllerTest {
private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
private TestingFlinkService flinkService;
- private Context context;
+ private Context<FlinkDeployment> context;
private TestingFlinkDeploymentController testController;
private KubernetesMockServer mockServer;
@@ -100,7 +100,7 @@ public class FlinkDeploymentControllerTest {
UpdateControl<FlinkDeployment> updateControl;
- FlinkDeployment appCluster = TestUtils.buildApplicationCluster(FlinkVersion.v1_16);
+ FlinkDeployment appCluster = TestUtils.buildApplicationCluster(flinkVersion);
assertEquals(
JobManagerDeploymentStatus.MISSING,
appCluster.getStatus().getJobManagerDeploymentStatus());
@@ -217,12 +217,13 @@ public class FlinkDeploymentControllerTest {
var submittedEventValidatingResponseProvider =
new TestUtils.ValidatingResponseProvider<>(
new EventBuilder().withNewMetadata().endMetadata().build(),
- r -> {
- assertTrue(
- r.getBody()
- .readUtf8()
- .contains(AbstractFlinkResourceReconciler.MSG_SUBMIT));
- });
+ r ->
+ assertTrue(
+ r.getBody()
+ .readUtf8()
+ .contains(
+ AbstractFlinkResourceReconciler
+ .MSG_SUBMIT)));
mockServer
.expect()
.post()
@@ -233,9 +234,11 @@ public class FlinkDeploymentControllerTest {
var validatingResponseProvider =
new TestUtils.ValidatingResponseProvider<>(
new EventBuilder().withNewMetadata().endMetadata().build(),
- r -> {
- assertTrue(r.getBody().readUtf8().contains(TestUtils.DEPLOYMENT_ERROR));
- });
+ r ->
+ assertTrue(
+ r.getBody()
+ .readUtf8()
+ .contains(TestUtils.DEPLOYMENT_ERROR)));
mockServer
.expect()
.post()
@@ -288,12 +291,13 @@ public class FlinkDeploymentControllerTest {
var submittedEventValidatingResponseProvider =
new TestUtils.ValidatingResponseProvider<>(
new EventBuilder().withNewMetadata().endMetadata().build(),
- r -> {
- assertTrue(
- r.getBody()
- .readUtf8()
- .contains(AbstractFlinkResourceReconciler.MSG_SUBMIT));
- });
+ r ->
+ assertTrue(
+ r.getBody()
+ .readUtf8()
+ .contains(
+ AbstractFlinkResourceReconciler
+ .MSG_SUBMIT)));
mockServer
.expect()
.post()
@@ -516,7 +520,7 @@ public class FlinkDeploymentControllerTest {
testController.reconcile(appCluster, context);
jobs = flinkService.listJobs();
assertEquals(1, jobs.size());
- assertEquals(null, jobs.get(0).f0);
+ assertNull(jobs.get(0).f0);
testController.reconcile(appCluster, context);
testController.reconcile(appCluster, context);
@@ -549,7 +553,7 @@ public class FlinkDeploymentControllerTest {
EventRecorder.Reason.valueOf(testController.events().poll().getReason()));
jobs = flinkService.listJobs();
assertEquals(1, jobs.size());
- assertEquals(null, jobs.get(0).f0);
+ assertNull(jobs.get(0).f0);
// Inject validation error in the middle of the upgrade
appCluster.getSpec().setRestartNonce(123L);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
index 92ead485..a6a57b4b 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
@@ -56,7 +56,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class RollbackTest {
private TestingFlinkService flinkService;
- private Context context;
+ private Context<FlinkDeployment> context;
private TestingFlinkDeploymentController testController;
@@ -193,7 +193,7 @@ public class RollbackTest {
@ParameterizedTest
@EnumSource(FlinkVersion.class)
public void testRollbackStateless(FlinkVersion flinkVersion) throws Exception {
- var dep = TestUtils.buildApplicationCluster();
+ var dep = TestUtils.buildApplicationCluster(flinkVersion);
dep.getSpec().getJob().setUpgradeMode(UpgradeMode.STATELESS);
testRollback(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
index b2991479..7a364979 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
@@ -63,7 +63,7 @@ public class TestingFlinkDeploymentController
private EventCollector eventCollector = new EventCollector();
private EventRecorder eventRecorder;
- private StatusRecorder statusRecorder;
+ private StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> statusRecorder;
public TestingFlinkDeploymentController(
FlinkConfigManager configManager,
@@ -137,16 +137,14 @@ public class TestingFlinkDeploymentController
}
private static class StatusUpdateCounter
- implements BiConsumer<
- AbstractFlinkResource<?, FlinkDeploymentStatus>, FlinkDeploymentStatus> {
+ implements BiConsumer<FlinkDeployment, FlinkDeploymentStatus> {
private FlinkDeployment currentResource;
private int counter;
@Override
public void accept(
- AbstractFlinkResource<?, FlinkDeploymentStatus>
- flinkDeploymentStatusAbstractFlinkResource,
+ FlinkDeployment flinkDeploymentStatusAbstractFlinkResource,
FlinkDeploymentStatus flinkDeploymentStatus) {
currentResource.setStatus(flinkDeploymentStatusAbstractFlinkResource.getStatus());
counter++;
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
index b02f112f..dc2f28d5 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
@@ -26,8 +26,8 @@ import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.Test;
import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR;
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
/** @link KubernetesOperatorMetricGroup tests. */
public class KubernetesOperatorMetricGroupTest {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetricsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetricsTest.java
index 5b0186db..7c49a1c4 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetricsTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetricsTest.java
@@ -62,9 +62,8 @@ public class OperatorJosdkMetricsTest {
TestingMetricRegistry.builder()
.setDelimiter(".".charAt(0))
.setRegisterConsumer(
- (metric, name, group) -> {
- metrics.put(group.getMetricIdentifier(name), metric);
- })
+ (metric, name, group) ->
+ metrics.put(group.getMetricIdentifier(name), metric))
.build();
operatorMetrics =
new OperatorJosdkMetrics(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
index 5182ffa9..f3baba7d 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
@@ -23,6 +23,8 @@ import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.TestingStatusRecorder;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
import org.apache.flink.kubernetes.operator.crd.status.SavepointInfo;
import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
@@ -49,14 +51,13 @@ public class SavepointObserverTest {
private KubernetesClient kubernetesClient;
private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
private final TestingFlinkService flinkService = new TestingFlinkService();
- private SavepointObserver observer;
- private final EventRecorder eventRecorder = new EventRecorder(null, (r, e) -> {});
+ private SavepointObserver<FlinkDeployment, FlinkDeploymentStatus> observer;
@BeforeEach
public void before() {
var eventRecorder = new EventRecorder(kubernetesClient, (r, e) -> {});
observer =
- new SavepointObserver(
+ new SavepointObserver<>(
flinkService, configManager, new TestingStatusRecorder<>(), eventRecorder);
}
@@ -132,7 +133,7 @@ public class SavepointObserverTest {
}
@Test
- public void testPeriodicSavepoint() throws Exception {
+ public void testPeriodicSavepoint() {
var conf = new Configuration();
var deployment = TestUtils.buildApplicationCluster();
deployment
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
index f3f93de6..19827276 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -49,8 +49,8 @@ import java.time.Instant;
import java.util.HashMap;
import java.util.stream.Collectors;
-import static org.junit.Assert.assertFalse;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -59,7 +59,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
@EnableKubernetesMockClient(crud = true)
public class ApplicationObserverTest {
private KubernetesClient kubernetesClient;
- private final Context readyContext = TestUtils.createContextWithReadyJobManagerDeployment();
+ private final Context<FlinkDeployment> readyContext =
+ TestUtils.createContextWithReadyJobManagerDeployment();
private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
private final TestingFlinkService flinkService = new TestingFlinkService();
private ApplicationObserver observer;
@@ -432,10 +433,10 @@ public class ApplicationObserverTest {
Exception exception =
assertThrows(
DeploymentFailedException.class,
- () -> {
- observer.observe(
- deployment, TestUtils.createContextWithInProgressDeployment());
- });
+ () ->
+ observer.observe(
+ deployment,
+ TestUtils.createContextWithInProgressDeployment()));
assertEquals(podFailedMessage, exception.getMessage());
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
index a9e06049..2d4acd51 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
@@ -39,7 +39,8 @@ import static org.junit.jupiter.api.Assertions.assertNull;
/** {@link SessionObserver} unit tests. */
public class SessionObserverTest {
- private final Context readyContext = TestUtils.createContextWithReadyJobManagerDeployment();
+ private final Context<FlinkDeployment> readyContext =
+ TestUtils.createContextWithReadyJobManagerDeployment();
private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
private final TestingFlinkService flinkService = new TestingFlinkService();
private SessionObserver observer;
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
index 4b00ed4f..f970bb66 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
@@ -214,10 +214,7 @@ public class SessionJobObserverTest {
});
// submit job
Assertions.assertThrows(
- RuntimeException.class,
- () -> {
- reconciler.reconcile(sessionJob, readyContext);
- });
+ RuntimeException.class, () -> reconciler.reconcile(sessionJob, readyContext));
Assertions.assertNotNull(sessionJob.getStatus().getReconciliationStatus());
Assertions.assertEquals(
ReconciliationState.UPGRADING,
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index 89f794e6..ac4272b1 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -57,11 +57,11 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.ThreadLocalRandom;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
/** @link JobStatusObserver unit tests */
@@ -70,14 +70,17 @@ public class ApplicationReconcilerTest {
private KubernetesClient kubernetesClient;
private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
- private TestingFlinkService flinkService = new TestingFlinkService();
+ private TestingFlinkService flinkService;
private ApplicationReconciler reconciler;
+ private Context<FlinkDeployment> context;
@BeforeEach
public void before() {
kubernetesClient.resource(TestUtils.buildApplicationCluster()).createOrReplace();
var eventRecorder = new EventRecorder(kubernetesClient, (r, e) -> {});
var statusRecoder = new TestingStatusRecorder<FlinkDeployment, FlinkDeploymentStatus>();
+ flinkService = new TestingFlinkService();
+ context = flinkService.getContext();
reconciler =
new ApplicationReconciler(
kubernetesClient,
@@ -90,8 +93,6 @@ public class ApplicationReconcilerTest {
@ParameterizedTest
@EnumSource(FlinkVersion.class)
public void testUpgrade(FlinkVersion flinkVersion) throws Exception {
- Context context = flinkService.getContext();
-
FlinkDeployment deployment = TestUtils.buildApplicationCluster(flinkVersion);
reconciler.reconcile(deployment, context);
@@ -104,7 +105,6 @@ public class ApplicationReconcilerTest {
statelessUpgrade.getSpec().getFlinkConfiguration().put("new", "conf");
reconciler.reconcile(statelessUpgrade, context);
- runningJobs = flinkService.listJobs();
assertEquals(0, flinkService.getRunningCount());
reconciler.reconcile(statelessUpgrade, context);
@@ -125,7 +125,6 @@ public class ApplicationReconcilerTest {
reconciler.reconcile(statefulUpgrade, context);
- runningJobs = flinkService.listJobs();
assertEquals(0, flinkService.getRunningCount());
reconciler.reconcile(statefulUpgrade, context);
@@ -184,8 +183,6 @@ public class ApplicationReconcilerTest {
@Test
public void testUpgradeModeChangeFromSavepointToLastState() throws Exception {
final String expectedSavepointPath = "savepoint_0";
- Context context = flinkService.getContext();
-
final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
@@ -224,7 +221,6 @@ public class ApplicationReconcilerTest {
@Test
public void triggerSavepoint() throws Exception {
- Context context = flinkService.getContext();
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
@@ -314,7 +310,6 @@ public class ApplicationReconcilerTest {
public void testUpgradeModeChangedToLastStateShouldTriggerSavepointWhileHADisabled()
throws Exception {
flinkService.setHaDataAvailable(false);
- Context context = flinkService.getContext();
final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
deployment.getSpec().getFlinkConfiguration().remove(HighAvailabilityOptions.HA_MODE.key());
@@ -359,8 +354,6 @@ public class ApplicationReconcilerTest {
@Test
public void testUpgradeModeChangedToLastStateShouldNotTriggerSavepointWhileHAEnabled()
throws Exception {
- Context context = flinkService.getContext();
-
final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
@@ -393,8 +386,6 @@ public class ApplicationReconcilerTest {
@Test
public void triggerRestart() throws Exception {
- Context context = flinkService.getContext();
-
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
@@ -452,7 +443,6 @@ public class ApplicationReconcilerTest {
@Test
public void testJobUpgradeIgnorePendingSavepoint() throws Exception {
- Context context = flinkService.getContext();
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
List<Tuple2<String, JobStatusMessage>> runningJobs = flinkService.listJobs();
@@ -488,8 +478,6 @@ public class ApplicationReconcilerTest {
@Test
public void testRandomJobResultStorePath() throws Exception {
FlinkDeployment flinkApp = TestUtils.buildApplicationCluster();
- Context context = flinkService.getContext();
-
final String haStoragePath = "file:///flink-data/ha";
flinkApp.getSpec()
.getFlinkConfiguration()
@@ -520,15 +508,14 @@ public class ApplicationReconcilerTest {
var deployment = TestUtils.buildApplicationCluster(FlinkVersion.v1_14);
deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
- var ctx = flinkService.getContext();
- reconciler.reconcile(deployment, ctx);
+ reconciler.reconcile(deployment, context);
deployment.getSpec().setFlinkVersion(FlinkVersion.v1_15);
var reconStatus = deployment.getStatus().getReconciliationStatus();
// Do not trigger update until running
- reconciler.reconcile(deployment, ctx);
+ reconciler.reconcile(deployment, context);
assertEquals(ReconciliationState.DEPLOYED, reconStatus.getState());
deployment.getStatus().getJobStatus().setState(JobState.RUNNING.name());
@@ -537,7 +524,7 @@ public class ApplicationReconcilerTest {
.getJobStatus()
.setJobId(flinkService.listJobs().get(0).f1.getJobId().toHexString());
- reconciler.reconcile(deployment, ctx);
+ reconciler.reconcile(deployment, context);
assertEquals(ReconciliationState.UPGRADING, reconStatus.getState());
assertEquals(
UpgradeMode.SAVEPOINT,
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
index 6fc4764e..6b1cf078 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
@@ -42,7 +42,6 @@ import org.apache.flink.runtime.client.JobStatusMessage;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -53,9 +52,9 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Tests for {@link SessionJobReconciler}. */
@@ -68,6 +67,7 @@ public class SessionJobReconcilerTest {
private EventRecorder eventRecorder;
private SessionJobReconciler reconciler;
private StatusRecorder<FlinkSessionJob, FlinkSessionJobStatus> statusRecoder;
+ private FlinkServiceFactory flinkServiceFactory;
@BeforeEach
public void before() {
@@ -85,19 +85,17 @@ public class SessionJobReconcilerTest {
return false;
}
};
- reconciler =
- new SessionJobReconciler(
- null, flinkServiceFactory, configManager, eventRecorder, statusRecoder);
- kubernetesClient.resource(TestUtils.buildSessionJob()).createOrReplace();
statusRecoder = new TestingStatusRecorder<>();
reconciler =
new SessionJobReconciler(
- null, flinkServiceFactory, configManager, eventRecorder, statusRecoder);
+ kubernetesClient,
+ flinkServiceFactory,
+ configManager,
+ eventRecorder,
+ statusRecoder);
+ kubernetesClient.resource(TestUtils.buildSessionJob()).createOrReplace();
}
- private FlinkServiceFactory flinkServiceFactory;
- private FlinkSessionJob sessionJob;
-
@Test
public void testSubmitAndCleanUp() throws Exception {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
@@ -437,7 +435,7 @@ public class SessionJobReconcilerTest {
@Test
public void testJobUpgradeIgnorePendingSavepoint() throws Exception {
- Context readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
reconciler.reconcile(sessionJob, readyContext);
verifyAndSetRunningJobsToStatus(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
index bbffabbd..ec09f546 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
@@ -24,7 +24,6 @@ import org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.utils.StandaloneKubernetesUtils;
-import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.api.model.apps.Deployment;
@@ -35,10 +34,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
/** @link StandaloneFlinkService unit tests */
@EnableKubernetesMockClient(crud = true)
@@ -57,10 +54,6 @@ public class StandaloneFlinkServiceTest {
kubernetesClient = mockServer.createClient().inAnyNamespace();
flinkStandaloneService =
new StandaloneFlinkService(kubernetesClient, new FlinkConfigManager(configuration));
-
- ExecutorService executorService =
- Executors.newFixedThreadPool(
- 1, new ExecutorThreadFactory("flink-kubeclient-io-for-standalone-service"));
}
@Test
@@ -102,12 +95,11 @@ public class StandaloneFlinkServiceTest {
private Configuration buildConfig(FlinkDeployment flinkDeployment, Configuration configuration)
throws Exception {
- return configuration =
- FlinkConfigBuilder.buildFrom(
- flinkDeployment.getMetadata().getNamespace(),
- flinkDeployment.getMetadata().getName(),
- flinkDeployment.getSpec(),
- configuration);
+ return FlinkConfigBuilder.buildFrom(
+ flinkDeployment.getMetadata().getNamespace(),
+ flinkDeployment.getMetadata().getName(),
+ flinkDeployment.getSpec(),
+ configuration);
}
private void createDeployments() {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkUtilsTest.java
index dae68fff..dc2ac584 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkUtilsTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkUtilsTest.java
@@ -36,8 +36,8 @@ import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.junit.jupiter.api.Test;
import java.net.HttpURLConnection;
-import java.util.Arrays;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -53,20 +53,17 @@ public class FlinkUtilsTest {
KubernetesMockServer mockServer;
@Test
- public void testMergePods() throws Exception {
-
+ public void testMergePods() {
Container container1 = new Container();
container1.setName("container1");
Container container2 = new Container();
container2.setName("container2");
- Pod pod1 =
- TestUtils.getTestPod(
- "pod1 hostname", "pod1 api version", Arrays.asList(container2));
+ Pod pod1 = TestUtils.getTestPod("pod1 hostname", "pod1 api version", List.of(container2));
Pod pod2 =
TestUtils.getTestPod(
- "pod2 hostname", "pod2 api version", Arrays.asList(container1, container2));
+ "pod2 hostname", "pod2 api version", List.of(container1, container2));
Pod mergedPod = FlinkUtils.mergePodTemplates(pod1, pod2);
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
index cee40710..70d8c9fd 100644
--- 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
@@ -36,7 +36,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.stream.Collectors;
-import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
/** Test class for {@link ValidatorUtils}. */
public class ValidatorUtilsTest {
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 23e75bc1..77a6c44b 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
@@ -158,15 +158,14 @@ public class DefaultValidatorTest {
CheckpointingOptions.SAVEPOINT_DIRECTORY.key()));
testError(
- dep -> {
- dep.getSpec()
- .setFlinkConfiguration(
- Map.of(
- KubernetesOperatorConfigOptions
- .PERIODIC_SAVEPOINT_INTERVAL
- .key(),
- "1m"));
- },
+ dep ->
+ dep.getSpec()
+ .setFlinkConfiguration(
+ Map.of(
+ KubernetesOperatorConfigOptions
+ .PERIODIC_SAVEPOINT_INTERVAL
+ .key(),
+ "1m")),
String.format(
"Periodic savepoints cannot be enabled when config key[%s] is not set",
CheckpointingOptions.SAVEPOINT_DIRECTORY.key()));
@@ -466,36 +465,33 @@ public class DefaultValidatorTest {
"The LAST_STATE upgrade mode is not supported in session job now.");
testSessionJobValidateWithModifier(
- sessionJob -> {
- sessionJob.getSpec().getJob().setState(JobState.SUSPENDED);
- },
+ sessionJob -> sessionJob.getSpec().getJob().setState(JobState.SUSPENDED),
flinkDeployment -> {},
"Job must start in running state");
testSessionJobValidateWithModifier(
- sessionJob -> {
- sessionJob
- .getSpec()
- .setFlinkConfiguration(
- Map.of(
- KubernetesOperatorConfigOptions.JAR_ARTIFACT_HTTP_HEADER
- .key(),
- "headerKey1:headerValue1,headerKey2:headerValue2"));
- },
+ sessionJob ->
+ sessionJob
+ .getSpec()
+ .setFlinkConfiguration(
+ Map.of(
+ KubernetesOperatorConfigOptions
+ .JAR_ARTIFACT_HTTP_HEADER
+ .key(),
+ "headerKey1:headerValue1,headerKey2:headerValue2")),
flinkDeployment -> {},
null);
testSessionJobValidateWithModifier(
- sessionJob -> {
- sessionJob
- .getSpec()
- .setFlinkConfiguration(
- Map.of(
- KubernetesOperatorConfigOptions
- .OPERATOR_RECONCILE_INTERVAL
- .key(),
- "60"));
- },
+ sessionJob ->
+ sessionJob
+ .getSpec()
+ .setFlinkConfiguration(
+ Map.of(
+ KubernetesOperatorConfigOptions
+ .OPERATOR_RECONCILE_INTERVAL
+ .key(),
+ "60")),
flinkDeployment -> {},
"Invalid session job flinkConfiguration key: kubernetes.operator.reconcile.interval."
+ " Allowed keys are [kubernetes.operator.user.artifacts.http.header]");