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 7d5bf953 [hotfix] Clean up code warnings and remove some unnecessary
bits + enforce junit5 assertions
7d5bf953 is described below
commit 7d5bf9536bdfbf86de5803766b28e503cd32ee04
Author: Gyula Fora <[email protected]>
AuthorDate: Wed Aug 3 09:26:45 2022 +0200
[hotfix] Clean up code warnings and remove some unnecessary bits + enforce
junit5 assertions
---
.../flink/kubernetes/operator/FlinkOperator.java | 6 +-
.../controller/FlinkSessionJobController.java | 2 +-
.../kubernetes/operator/crd/spec/FlinkVersion.java | 18 ------
.../operator/crd/status/ReconciliationState.java | 2 +-
.../operator/docs/CrdReferenceDoclet.java | 16 ++---
.../operator/metrics/OperatorJosdkMetrics.java | 1 -
.../operator/metrics/OperatorMetricUtils.java | 4 --
.../metrics/lifecycle/LifecycleMetrics.java | 70 +++++++++++-----------
.../kubernetes/operator/observer/Observer.java | 2 +-
.../operator/observer/SavepointObserver.java | 7 +--
.../context/ApplicationObserverContext.java | 4 +-
.../deployment/AbstractDeploymentObserver.java | 11 ++--
.../observer/deployment/ApplicationObserver.java | 6 +-
.../observer/deployment/ObserverFactory.java | 1 -
.../observer/deployment/SessionObserver.java | 2 +-
.../observer/sessionjob/SessionJobObserver.java | 12 ++--
.../kubernetes/operator/reconciler/Reconciler.java | 4 +-
.../operator/reconciler/ReconciliationUtils.java | 11 ++--
.../AbstractFlinkResourceReconciler.java | 28 ++++-----
.../deployment/AbstractJobReconciler.java | 14 ++---
.../deployment/ApplicationReconciler.java | 18 +++---
.../reconciler/deployment/SessionReconciler.java | 19 +++---
.../sessionjob/SessionJobReconciler.java | 14 ++---
.../operator/service/AbstractFlinkService.java | 4 +-
.../kubernetes/operator/utils/EventRecorder.java | 4 --
.../operator/validation/DefaultValidator.java | 1 +
.../operator/observer/SavepointObserverTest.java | 5 +-
.../deployment/ApplicationObserverTest.java | 5 +-
.../sessionjob/SessionJobObserverTest.java | 4 +-
.../StandaloneKubernetesJobManagerParameters.java | 3 +-
.../KubernetesStandaloneClusterDescriptor.java | 6 +-
.../operator/utils/StandaloneKubernetesUtils.java | 8 +--
.../StandaloneKubernetesJobManagerFactoryTest.java | 2 +-
...ndaloneKubernetesTaskManagerParametersTest.java | 9 +--
.../operator/kubeclient/utils/TestUtils.java | 2 +-
.../KubernetesStandaloneClusterDescriptorTest.java | 11 ++--
.../operator/admission/AdmissionHandler.java | 2 +-
.../operator/admission/FlinkOperatorWebhook.java | 2 +-
.../admission/informer/InformerManager.java | 12 ++--
.../admission/mutator/DefaultRequestMutator.java | 5 +-
.../admission/informer/InformerManagerTest.java | 2 +-
tools/maven/checkstyle.xml | 5 ++
42 files changed, 149 insertions(+), 215 deletions(-)
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 c0e8841d..07e8e98e 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
@@ -69,7 +69,7 @@ public class FlinkOperator {
private final FlinkServiceFactory flinkServiceFactory;
private final FlinkConfigManager configManager;
private final Set<FlinkResourceValidator> validators;
- @VisibleForTesting final Set<RegisteredController> registeredControllers =
new HashSet<>();
+ @VisibleForTesting final Set<RegisteredController<?>>
registeredControllers = new HashSet<>();
private final KubernetesOperatorMetricGroup metricGroup;
private final Collection<FlinkResourceListener> listeners;
@@ -149,9 +149,7 @@ public class FlinkOperator {
var reconciler =
new SessionJobReconciler(
client, flinkServiceFactory, configManager,
eventRecorder, statusRecorder);
- var observer =
- new SessionJobObserver(
- flinkServiceFactory, configManager, statusRecorder,
eventRecorder);
+ var observer = new SessionJobObserver(flinkServiceFactory,
configManager, eventRecorder);
var controller =
new FlinkSessionJobController(
configManager, validators, reconciler, observer,
statusRecorder);
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 aa4c1450..6d23f798 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
@@ -122,7 +122,7 @@ public class FlinkSessionJobController
EventSourceUtils.getFlinkDeploymentInformerEventSource(context));
}
- private boolean validateSessionJob(FlinkSessionJob sessionJob, Context
context) {
+ private boolean validateSessionJob(FlinkSessionJob sessionJob, Context<?>
context) {
for (FlinkResourceValidator validator : validators) {
Optional<String> validationError =
validator.validateSessionJob(
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkVersion.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkVersion.java
index dbe95724..924319f8 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkVersion.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkVersion.java
@@ -20,11 +20,6 @@ package org.apache.flink.kubernetes.operator.crd.spec;
import org.apache.flink.annotation.Experimental;
-import java.util.LinkedHashSet;
-import java.util.Set;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
/** Enumeration for supported Flink versions. */
@Experimental
public enum FlinkVersion {
@@ -37,19 +32,6 @@ public enum FlinkVersion {
return this.ordinal() > otherVersion.ordinal();
}
- /**
- * Returns all versions within the defined range, inclusive both start and
end.
- *
- * @param start Starting version.
- * @param end Last version.
- * @return Versions within the range.
- */
- public static Set<FlinkVersion> rangeOf(FlinkVersion start, FlinkVersion
end) {
- return Stream.of(FlinkVersion.values())
- .filter(v -> v.ordinal() >= start.ordinal() && v.ordinal() <=
end.ordinal())
- .collect(Collectors.toCollection(LinkedHashSet::new));
- }
-
/**
* Returns the current version.
*
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationState.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationState.java
index 13e2ce86..6d5e6eec 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationState.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationState.java
@@ -26,5 +26,5 @@ public enum ReconciliationState {
/** In the process of rolling back to the lastStableSpec. */
ROLLING_BACK,
/** Rolled back to the lastStableSpec. */
- ROLLED_BACK;
+ ROLLED_BACK
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/docs/CrdReferenceDoclet.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/docs/CrdReferenceDoclet.java
index 29fe798d..7bd418d8 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/docs/CrdReferenceDoclet.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/docs/CrdReferenceDoclet.java
@@ -93,7 +93,7 @@ public class CrdReferenceDoclet implements Doclet {
printStream.write(FileUtils.readFileToByteArray(new
File(templateFile)));
MdPrinter se = new MdPrinter(printStream);
- printStream.println("");
+ printStream.println();
printStream.println("## Spec");
var spec =
sortedByName(
@@ -103,7 +103,7 @@ public class CrdReferenceDoclet implements Doclet {
handleAbstractClass(spec, environment.getTypeUtils());
se.show(spec);
- printStream.println("");
+ printStream.println();
printStream.println("## Status");
var status =
sortedByName(
@@ -173,12 +173,12 @@ public class CrdReferenceDoclet implements Doclet {
ElementKind kind = e.getKind();
switch (kind) {
case CLASS:
- out.println("");
+ out.println();
out.println("### " + e.getSimpleName());
out.println("**Class**: " + e);
- out.println("");
+ out.println();
out.println("**Description**: " + dcTree);
- out.println("");
+ out.println();
out.println("| Parameter | Type | Docs |");
out.println("| ----------| ---- | ---- |");
// if this is a child class, print it's parent's enclosed
elements.
@@ -198,12 +198,12 @@ public class CrdReferenceDoclet implements Doclet {
+ " |");
return null;
case ENUM:
- out.println("");
+ out.println();
out.println("### " + e.getSimpleName());
out.println("**Class**: " + e);
- out.println("");
+ out.println();
out.println("**Description**: " + dcTree);
- out.println("");
+ out.println();
out.println("| Value | Docs |");
out.println("| ----- | ---- |");
break;
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
index 6b880e8c..fa09fb97 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
@@ -84,7 +84,6 @@ public class OperatorJosdkMetrics implements Metrics {
histogram(execution,
execution.successTypeName(result)).update(toSeconds(startTime));
return result;
} catch (Exception e) {
- var h = histogram(execution, "failed");
histogram(execution, "failed").update(toSeconds(startTime));
throw e;
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
index 6b989907..c5f6be75 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
@@ -98,10 +98,6 @@ public class OperatorMetricUtils {
ReporterSetup.fromConfiguration(configuration, pluginManager));
}
- public static Histogram synchronizedHistogram(Histogram histogram) {
- return new SynchronizedHistogram(histogram);
- }
-
public static Counter synchronizedCounter(Counter counter) {
return new SynchronizedCounter(counter);
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java
index ad76b02a..872e749c 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java
@@ -177,48 +177,46 @@ public class LifecycleMetrics<CR extends
AbstractFlinkResource<?, ?>>
private Map<String, List<Histogram>> getTransitionHistograms(CR cr) {
var histos = new HashMap<String, List<Histogram>>();
transitionMetrics.forEach(
- (metricName, t) -> {
- histos.put(
- metricName,
- namespaceHistosEnabled
- ? List.of(
- t.f0,
- t.f1.computeIfAbsent(
-
cr.getMetadata().getNamespace(),
- ns ->
-
createTransitionHistogram(
- metricName,
-
operatorMetricGroup
-
.createResourceNamespaceGroup(
-
configManager
-
.getDefaultConfig(),
-
ns))))
- : List.of(t.f0));
- });
+ (metricName, t) ->
+ histos.put(
+ metricName,
+ namespaceHistosEnabled
+ ? List.of(
+ t.f0,
+ t.f1.computeIfAbsent(
+
cr.getMetadata().getNamespace(),
+ ns ->
+
createTransitionHistogram(
+
metricName,
+
operatorMetricGroup
+
.createResourceNamespaceGroup(
+
configManager
+
.getDefaultConfig(),
+
ns))))
+ : List.of(t.f0)));
return histos;
}
private Map<ResourceLifecycleState, List<Histogram>>
getStateTimeHistograms(CR cr) {
var histos = new HashMap<ResourceLifecycleState, List<Histogram>>();
stateTimeMetrics.forEach(
- (state, t) -> {
- histos.put(
- state,
- namespaceHistosEnabled
- ? List.of(
- t.f0,
- t.f1.computeIfAbsent(
-
cr.getMetadata().getNamespace(),
- ns ->
-
createStateTimeHistogram(
- state,
-
operatorMetricGroup
-
.createResourceNamespaceGroup(
-
configManager
-
.getDefaultConfig(),
-
ns))))
- : List.of(t.f0));
- });
+ (state, t) ->
+ histos.put(
+ state,
+ namespaceHistosEnabled
+ ? List.of(
+ t.f0,
+ t.f1.computeIfAbsent(
+
cr.getMetadata().getNamespace(),
+ ns ->
+
createStateTimeHistogram(
+ state,
+
operatorMetricGroup
+
.createResourceNamespaceGroup(
+
configManager
+
.getDefaultConfig(),
+
ns))))
+ : List.of(t.f0)));
return histos;
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/Observer.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/Observer.java
index 29621ea6..2e1ef358 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/Observer.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/Observer.java
@@ -28,5 +28,5 @@ public interface Observer<CR> {
* @param cr the target custom resource
* @param context the context with which the operation is executed
*/
- void observe(CR cr, Context context);
+ void observe(CR cr, Context<?> context);
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
index 52c322ee..091badfc 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
@@ -32,7 +32,6 @@ import
org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.ConfigOptionUtils;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
-import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,22 +48,18 @@ public class SavepointObserver<
private final FlinkService flinkService;
private final FlinkConfigManager configManager;
- private final StatusRecorder<CR, STATUS> statusRecorder;
private final EventRecorder eventRecorder;
public SavepointObserver(
FlinkService flinkService,
FlinkConfigManager configManager,
- StatusRecorder<CR, STATUS> statusRecorder,
EventRecorder eventRecorder) {
this.flinkService = flinkService;
this.configManager = configManager;
- this.statusRecorder = statusRecorder;
this.eventRecorder = eventRecorder;
}
- public void observeSavepointStatus(
- AbstractFlinkResource<?, STATUS> resource, Configuration
deployedConfig) {
+ public void observeSavepointStatus(CR resource, Configuration
deployedConfig) {
var jobStatus = resource.getStatus().getJobStatus();
var savepointInfo = jobStatus.getSavepointInfo();
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/ApplicationObserverContext.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/ApplicationObserverContext.java
index 3c2351ee..6582d792 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/ApplicationObserverContext.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/ApplicationObserverContext.java
@@ -26,11 +26,11 @@ import io.javaoperatorsdk.operator.api.reconciler.Context;
public class ApplicationObserverContext {
public final FlinkDeployment flinkApp;
- public final Context context;
+ public final Context<?> context;
public final Configuration deployedConfig;
public ApplicationObserverContext(
- FlinkDeployment flinkApp, Context context, Configuration
deployedConfig) {
+ FlinkDeployment flinkApp, Context<?> context, Configuration
deployedConfig) {
this.flinkApp = flinkApp;
this.context = context;
this.deployedConfig = deployedConfig;
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
index e3b5ebd6..0ff0ce57 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
@@ -27,7 +27,6 @@ import org.apache.flink.kubernetes.operator.crd.spec.JobState;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
-import org.apache.flink.kubernetes.operator.crd.status.ReconciliationStatus;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import org.apache.flink.kubernetes.operator.observer.Observer;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
@@ -72,7 +71,7 @@ public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeploy
}
@Override
- public void observe(FlinkDeployment flinkApp, Context context) {
+ public void observe(FlinkDeployment flinkApp, Context<?> context) {
var status = flinkApp.getStatus();
var reconciliationStatus = status.getReconciliationStatus();
@@ -119,7 +118,7 @@ public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeploy
}
protected void observeJmDeployment(
- FlinkDeployment flinkApp, Context context, Configuration
effectiveConfig) {
+ FlinkDeployment flinkApp, Context<?> context, Configuration
effectiveConfig) {
FlinkDeploymentStatus deploymentStatus = flinkApp.getStatus();
JobManagerDeploymentStatus previousJmStatus =
deploymentStatus.getJobManagerDeploymentStatus();
@@ -215,7 +214,7 @@ public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeploy
protected void clearErrorsIfDeploymentIsHealthy(FlinkDeployment dep) {
FlinkDeploymentStatus status = dep.getStatus();
- ReconciliationStatus reconciliationStatus =
status.getReconciliationStatus();
+ var reconciliationStatus = status.getReconciliationStatus();
if (status.getJobManagerDeploymentStatus() !=
JobManagerDeploymentStatus.ERROR
&&
!JobStatus.FAILED.name().equals(dep.getStatus().getJobStatus().getState())
&& reconciliationStatus.isLastReconciledSpecStable()) {
@@ -261,7 +260,7 @@ public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeploy
* @param flinkDep Flink resource to check.
* @param context Context for reconciliation.
*/
- private void checkIfAlreadyUpgraded(FlinkDeployment flinkDep, Context
context) {
+ private void checkIfAlreadyUpgraded(FlinkDeployment flinkDep, Context<?>
context) {
var status = flinkDep.getStatus();
if (status.getReconciliationStatus().isFirstDeployment()) {
return;
@@ -309,5 +308,5 @@ public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeploy
* @param deployedConfig config that is deployed on the Flink cluster
*/
protected abstract void observeFlinkCluster(
- FlinkDeployment flinkApp, Context context, Configuration
deployedConfig);
+ FlinkDeployment flinkApp, Context<?> context, Configuration
deployedConfig);
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
index 038e0342..dd658b83 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
@@ -27,7 +27,6 @@ import
org.apache.flink.kubernetes.operator.observer.SavepointObserver;
import
org.apache.flink.kubernetes.operator.observer.context.ApplicationObserverContext;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
-import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.runtime.client.JobStatusMessage;
import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -44,11 +43,10 @@ public class ApplicationObserver extends
AbstractDeploymentObserver {
public ApplicationObserver(
FlinkService flinkService,
FlinkConfigManager configManager,
- StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>
statusRecorder,
EventRecorder eventRecorder) {
super(flinkService, configManager, eventRecorder);
this.savepointObserver =
- new SavepointObserver(flinkService, configManager,
statusRecorder, eventRecorder);
+ new SavepointObserver<>(flinkService, configManager,
eventRecorder);
this.jobStatusObserver =
new JobStatusObserver<>(flinkService, eventRecorder) {
@Override
@@ -69,7 +67,7 @@ public class ApplicationObserver extends
AbstractDeploymentObserver {
@Override
protected void observeFlinkCluster(
- FlinkDeployment flinkApp, Context context, Configuration
deployedConfig) {
+ FlinkDeployment flinkApp, Context<?> context, Configuration
deployedConfig) {
boolean jobFound =
jobStatusObserver.observe(
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ObserverFactory.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ObserverFactory.java
index 1f1617b2..b2145622 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ObserverFactory.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ObserverFactory.java
@@ -69,7 +69,6 @@ public class ObserverFactory {
return new ApplicationObserver(
flinkServiceFactory.getOrCreate(flinkApp),
configManager,
- statusRecorder,
eventRecorder);
default:
throw new UnsupportedOperationException(
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserver.java
index 3debfd9b..f8f0010a 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserver.java
@@ -40,7 +40,7 @@ public class SessionObserver extends
AbstractDeploymentObserver {
@Override
public void observeFlinkCluster(
- FlinkDeployment deployment, Context context, Configuration
deployedConfig) {
+ FlinkDeployment deployment, Context<?> context, Configuration
deployedConfig) {
// Check if session cluster can serve rest calls following our
practice in JobObserver
try {
flinkService.listJobs(deployedConfig);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
index 14f76492..633744a1 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
@@ -35,7 +35,6 @@ import
org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.service.FlinkServiceFactory;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
-import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.util.Preconditions;
@@ -56,17 +55,14 @@ public class SessionJobObserver implements
Observer<FlinkSessionJob> {
private final FlinkServiceFactory flinkServiceFactory;
private final FlinkConfigManager configManager;
private final EventRecorder eventRecorder;
- private final StatusRecorder<FlinkSessionJob, FlinkSessionJobStatus>
statusRecorder;
public SessionJobObserver(
FlinkServiceFactory flinkServiceFactory,
FlinkConfigManager configManager,
- StatusRecorder<FlinkSessionJob, FlinkSessionJobStatus>
statusRecorder,
EventRecorder eventRecorder) {
this.flinkServiceFactory = flinkServiceFactory;
this.configManager = configManager;
this.eventRecorder = eventRecorder;
- this.statusRecorder = statusRecorder;
}
private JobStatusObserver<VoidObserverContext>
getJobStatusObserver(FlinkService flinkService) {
@@ -101,7 +97,7 @@ public class SessionJobObserver implements
Observer<FlinkSessionJob> {
}
@Override
- public void observe(FlinkSessionJob flinkSessionJob, Context context) {
+ public void observe(FlinkSessionJob flinkSessionJob, Context<?> context) {
if
(flinkSessionJob.getStatus().getReconciliationStatus().isFirstDeployment()) {
return;
}
@@ -129,9 +125,9 @@ public class SessionJobObserver implements
Observer<FlinkSessionJob> {
flinkSessionJob, deployedConfig,
VoidObserverContext.INSTANCE);
if (jobFound) {
- SavepointObserver savepointObserver =
- new SavepointObserver<>(
- flinkService, configManager, statusRecorder,
eventRecorder);
+ var savepointObserver =
+ new SavepointObserver<FlinkSessionJob,
FlinkSessionJobStatus>(
+ flinkService, configManager, eventRecorder);
savepointObserver.observeSavepointStatus(flinkSessionJob,
deployedConfig);
}
SavepointUtils.resetTriggerIfJobNotRunning(flinkSessionJob,
eventRecorder);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/Reconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/Reconciler.java
index 72bf7ae2..08026092 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/Reconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/Reconciler.java
@@ -35,7 +35,7 @@ public interface Reconciler<CR> {
* @param context the context with which the operation is executed
* @throws Exception Error during reconciliation.
*/
- void reconcile(CR cr, Context context) throws Exception;
+ void reconcile(CR cr, Context<?> context) throws Exception;
/**
* This is called when receiving the delete event of custom resource. This
method is meant to
@@ -45,5 +45,5 @@ public interface Reconciler<CR> {
* @param context the context with which the operation is executed
* @return DeleteControl to manage the deletion behavior
*/
- DeleteControl cleanup(CR cr, Context context);
+ DeleteControl cleanup(CR cr, Context<?> context);
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
index 1e596fe5..70606512 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
@@ -402,12 +402,11 @@ public class ReconciliationUtils {
StatusRecorder<R, STATUS> statusRecorder) {
retryInfo.ifPresent(
- r -> {
- LOG.warn(
- "Attempt count: {}, last attempt: {}",
- r.getAttemptCount(),
- r.isLastAttempt());
- });
+ r ->
+ LOG.warn(
+ "Attempt count: {}, last attempt: {}",
+ r.getAttemptCount(),
+ r.isLastAttempt()));
statusRecorder.updateStatusFromCache(resource);
ReconciliationUtils.updateForReconciliationError(
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
index f2eb5efb..c51ba32c 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
@@ -83,7 +83,7 @@ public abstract class AbstractFlinkResourceReconciler<
}
@Override
- public final void reconcile(CR cr, Context ctx) throws Exception {
+ public final void reconcile(CR cr, Context<?> ctx) throws Exception {
var spec = cr.getSpec();
var deployConfig = getDeployConfig(cr.getMetadata(), spec, ctx);
var status = cr.getStatus();
@@ -167,7 +167,7 @@ public abstract class AbstractFlinkResourceReconciler<
* @param ctx Reconciliation context.
* @return Deployment configuration.
*/
- protected abstract Configuration getDeployConfig(ObjectMeta meta, SPEC
spec, Context ctx);
+ protected abstract Configuration getDeployConfig(ObjectMeta meta, SPEC
spec, Context<?> ctx);
/**
* Get Flink configuration for client interactions with the running Flink
deployment/session
@@ -177,7 +177,7 @@ public abstract class AbstractFlinkResourceReconciler<
* @param context Reconciliation context.
* @return Observe configuration.
*/
- protected abstract Configuration getObserveConfig(CR resource, Context
context);
+ protected abstract Configuration getObserveConfig(CR resource, Context<?>
context);
/**
* Check whether the given Flink resource is ready to be reconciled or we
are still waiting for
@@ -188,7 +188,7 @@ public abstract class AbstractFlinkResourceReconciler<
* @param deployConfig Deployment configuration.
* @return True if the resource is ready to be reconciled.
*/
- protected abstract boolean readyToReconcile(CR cr, Context ctx,
Configuration deployConfig);
+ protected abstract boolean readyToReconcile(CR cr, Context<?> ctx,
Configuration deployConfig);
/**
* Reconcile spec upgrade on the currently deployed/suspended Flink
resource and update the
@@ -200,7 +200,7 @@ public abstract class AbstractFlinkResourceReconciler<
* @throws Exception Error during spec upgrade.
*/
protected abstract void reconcileSpecChange(
- CR cr, Context ctx, Configuration observeConfig, Configuration
deployConfig)
+ CR cr, Context<?> ctx, Configuration observeConfig, Configuration
deployConfig)
throws Exception;
/**
@@ -211,7 +211,7 @@ public abstract class AbstractFlinkResourceReconciler<
* @param observeConfig Observe configuration.
* @throws Exception Error during rollback.
*/
- protected abstract void rollback(CR cr, Context ctx, Configuration
observeConfig)
+ protected abstract void rollback(CR cr, Context<?> ctx, Configuration
observeConfig)
throws Exception;
/**
@@ -224,10 +224,10 @@ public abstract class AbstractFlinkResourceReconciler<
* @throws Exception Error during reconciliation.
*/
protected abstract boolean reconcileOtherChanges(
- CR cr, Context context, Configuration observeConfig) throws
Exception;
+ CR cr, Context<?> context, Configuration observeConfig) throws
Exception;
@Override
- public final DeleteControl cleanup(CR resource, Context context) {
+ public final DeleteControl cleanup(CR resource, Context<?> context) {
return cleanupInternal(resource, context);
}
@@ -247,7 +247,7 @@ public abstract class AbstractFlinkResourceReconciler<
CR relatedResource,
SPEC spec,
STATUS status,
- Context ctx,
+ Context<?> ctx,
Configuration deployConfig,
Optional<String> savepoint,
boolean requireHaMetadata)
@@ -260,16 +260,16 @@ public abstract class AbstractFlinkResourceReconciler<
* @param context Current context.
* @return DeleteControl object.
*/
- protected abstract DeleteControl cleanupInternal(CR resource, Context
context);
+ protected abstract DeleteControl cleanupInternal(CR resource, Context<?>
context);
/**
* Get the Flink service related to the resource and context.
*
- * @param resource
- * @param context
- * @return
+ * @param resource Resource being reconciled.
+ * @param context Current context.
+ * @return Flink service implementation.
*/
- protected abstract FlinkService getFlinkService(CR resource, Context
context);
+ protected abstract FlinkService getFlinkService(CR resource, Context<?>
context);
/**
* Checks whether the desired spec already matches the currently deployed
spec. If they match
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
index 5d47ebbc..fea52829 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
@@ -60,7 +60,7 @@ public abstract class AbstractJobReconciler<
}
@Override
- public boolean readyToReconcile(CR resource, Context context,
Configuration deployConfig) {
+ public boolean readyToReconcile(CR resource, Context<?> context,
Configuration deployConfig) {
if (shouldWaitForPendingSavepoint(
resource.getStatus().getJobStatus(),
getDeployConfig(resource.getMetadata(), resource.getSpec(),
context))) {
@@ -78,7 +78,7 @@ public abstract class AbstractJobReconciler<
@Override
protected void reconcileSpecChange(
- CR resource, Context ctx, Configuration observeConfig,
Configuration deployConfig)
+ CR resource, Context<?> ctx, Configuration observeConfig,
Configuration deployConfig)
throws Exception {
STATUS status = resource.getStatus();
var reconciliationStatus = status.getReconciliationStatus();
@@ -173,7 +173,7 @@ public abstract class AbstractJobReconciler<
CR resource,
SPEC spec,
STATUS status,
- Context ctx,
+ Context<?> ctx,
Configuration deployConfig,
boolean requireHaMetadata)
throws Exception {
@@ -189,7 +189,7 @@ public abstract class AbstractJobReconciler<
}
@Override
- protected void rollback(CR resource, Context ctx, Configuration
observeConfig)
+ protected void rollback(CR resource, Context<?> ctx, Configuration
observeConfig)
throws Exception {
var reconciliationStatus =
resource.getStatus().getReconciliationStatus();
var rollbackSpec = reconciliationStatus.deserializeLastStableSpec();
@@ -216,8 +216,8 @@ public abstract class AbstractJobReconciler<
}
@Override
- public boolean reconcileOtherChanges(CR resource, Context context,
Configuration observeConfig)
- throws Exception {
+ public boolean reconcileOtherChanges(
+ CR resource, Context<?> context, Configuration observeConfig)
throws Exception {
return SavepointUtils.triggerSavepointIfNeeded(
getFlinkService(resource, context), resource, observeConfig);
}
@@ -231,6 +231,6 @@ public abstract class AbstractJobReconciler<
* @throws Exception Error during cancellation.
*/
protected abstract void cancelJob(
- CR resource, Context ctx, UpgradeMode upgradeMode, Configuration
observeConfig)
+ CR resource, Context<?> ctx, UpgradeMode upgradeMode,
Configuration observeConfig)
throws Exception;
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index e7c5143a..42aabaad 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -66,18 +66,18 @@ public class ApplicationReconciler
}
@Override
- protected FlinkService getFlinkService(FlinkDeployment resource, Context
context) {
+ protected FlinkService getFlinkService(FlinkDeployment resource,
Context<?> context) {
return flinkService;
}
@Override
- protected Configuration getObserveConfig(FlinkDeployment deployment,
Context context) {
+ protected Configuration getObserveConfig(FlinkDeployment deployment,
Context<?> context) {
return configManager.getObserveConfig(deployment);
}
@Override
protected Configuration getDeployConfig(
- ObjectMeta deployMeta, FlinkDeploymentSpec currentDeploySpec,
Context context) {
+ ObjectMeta deployMeta, FlinkDeploymentSpec currentDeploySpec,
Context<?> context) {
return configManager.getDeployConfig(deployMeta, currentDeploySpec);
}
@@ -125,7 +125,7 @@ public class ApplicationReconciler
FlinkDeployment relatedResource,
FlinkDeploymentSpec spec,
FlinkDeploymentStatus status,
- Context ctx,
+ Context<?> ctx,
Configuration deployConfig,
Optional<String> savepoint,
boolean requireHaMetadata)
@@ -165,7 +165,7 @@ public class ApplicationReconciler
@Override
protected void cancelJob(
FlinkDeployment deployment,
- Context ctx,
+ Context<?> ctx,
UpgradeMode upgradeMode,
Configuration observeConfig)
throws Exception {
@@ -192,7 +192,8 @@ public class ApplicationReconciler
@Override
public boolean reconcileOtherChanges(
- FlinkDeployment deployment, Context ctx, Configuration
observeConfig) throws Exception {
+ FlinkDeployment deployment, Context<?> ctx, Configuration
observeConfig)
+ throws Exception {
if (super.reconcileOtherChanges(deployment, ctx, observeConfig)) {
return true;
}
@@ -205,7 +206,8 @@ public class ApplicationReconciler
}
private void recoverJmDeployment(
- FlinkDeployment deployment, Context ctx, Configuration
observeConfig) throws Exception {
+ FlinkDeployment deployment, Context<?> ctx, Configuration
observeConfig)
+ throws Exception {
LOG.info("Missing Flink Cluster deployment, trying to recover...");
FlinkDeploymentSpec specToRecover =
ReconciliationUtils.getDeployedSpec(deployment);
restoreJob(deployment, specToRecover, deployment.getStatus(), ctx,
observeConfig, true);
@@ -213,7 +215,7 @@ public class ApplicationReconciler
@Override
@SneakyThrows
- protected DeleteControl cleanupInternal(FlinkDeployment deployment,
Context context) {
+ protected DeleteControl cleanupInternal(FlinkDeployment deployment,
Context<?> context) {
var status = deployment.getStatus();
if (status.getReconciliationStatus().isFirstDeployment()) {
flinkService.deleteClusterDeployment(deployment.getMetadata(),
status, true);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
index 9b2f48e1..2e2efcf6 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
@@ -66,31 +66,31 @@ public class SessionReconciler
}
@Override
- protected FlinkService getFlinkService(FlinkDeployment resource, Context
context) {
+ protected FlinkService getFlinkService(FlinkDeployment resource,
Context<?> context) {
return flinkService;
}
@Override
protected Configuration getDeployConfig(
- ObjectMeta meta, FlinkDeploymentSpec spec, Context ctx) {
+ ObjectMeta meta, FlinkDeploymentSpec spec, Context<?> ctx) {
return configManager.getDeployConfig(meta, spec);
}
@Override
- protected Configuration getObserveConfig(FlinkDeployment resource, Context
context) {
+ protected Configuration getObserveConfig(FlinkDeployment resource,
Context<?> context) {
return configManager.getObserveConfig(resource);
}
@Override
protected boolean readyToReconcile(
- FlinkDeployment deployment, Context ctx, Configuration
deployConfig) {
+ FlinkDeployment deployment, Context<?> ctx, Configuration
deployConfig) {
return true;
}
@Override
protected void reconcileSpecChange(
FlinkDeployment deployment,
- Context ctx,
+ Context<?> ctx,
Configuration observeConfig,
Configuration deployConfig)
throws Exception {
@@ -122,7 +122,7 @@ public class SessionReconciler
FlinkDeployment cr,
FlinkDeploymentSpec spec,
FlinkDeploymentStatus status,
- Context ctx,
+ Context<?> ctx,
Configuration deployConfig,
Optional<String> savepoint,
boolean requireHaMetadata)
@@ -133,7 +133,7 @@ public class SessionReconciler
}
@Override
- protected void rollback(FlinkDeployment deployment, Context ctx,
Configuration observeConfig)
+ protected void rollback(FlinkDeployment deployment, Context<?> ctx,
Configuration observeConfig)
throws Exception {
FlinkDeploymentStatus status = deployment.getStatus();
ReconciliationStatus<FlinkDeploymentSpec> reconciliationStatus =
@@ -157,7 +157,8 @@ public class SessionReconciler
@Override
public boolean reconcileOtherChanges(
- FlinkDeployment flinkApp, Context ctx, Configuration
observeConfig) throws Exception {
+ FlinkDeployment flinkApp, Context<?> ctx, Configuration
observeConfig)
+ throws Exception {
if (shouldRecoverDeployment(observeConfig, flinkApp)) {
recoverSession(flinkApp, observeConfig);
return true;
@@ -172,7 +173,7 @@ public class SessionReconciler
}
@Override
- public DeleteControl cleanupInternal(FlinkDeployment deployment, Context
context) {
+ public DeleteControl cleanupInternal(FlinkDeployment deployment,
Context<?> context) {
Set<FlinkSessionJob> sessionJobs =
context.getSecondaryResources(FlinkSessionJob.class);
if (!sessionJobs.isEmpty()) {
var error =
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
index 1ff4f76c..983095d0 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
@@ -59,7 +59,7 @@ public class SessionJobReconciler
}
@Override
- protected FlinkService getFlinkService(FlinkSessionJob resource, Context
context) {
+ protected FlinkService getFlinkService(FlinkSessionJob resource,
Context<?> context) {
Optional<FlinkDeployment> deploymentOpt =
context.getSecondaryResource(FlinkDeployment.class);
@@ -70,13 +70,13 @@ public class SessionJobReconciler
}
@Override
- protected Configuration getObserveConfig(FlinkSessionJob sessionJob,
Context context) {
+ protected Configuration getObserveConfig(FlinkSessionJob sessionJob,
Context<?> context) {
return getDeployConfig(sessionJob.getMetadata(), sessionJob.getSpec(),
context);
}
@Override
protected Configuration getDeployConfig(
- ObjectMeta deployMeta, FlinkSessionJobSpec currentDeploySpec,
Context context) {
+ ObjectMeta deployMeta, FlinkSessionJobSpec currentDeploySpec,
Context<?> context) {
Optional<FlinkDeployment> deploymentOpt =
context.getSecondaryResource(FlinkDeployment.class);
@@ -88,7 +88,7 @@ public class SessionJobReconciler
@Override
public boolean readyToReconcile(
- FlinkSessionJob flinkSessionJob, Context context, Configuration
deployConfig) {
+ FlinkSessionJob flinkSessionJob, Context<?> context, Configuration
deployConfig) {
return
sessionClusterReady(context.getSecondaryResource(FlinkDeployment.class))
&& super.readyToReconcile(flinkSessionJob, context,
deployConfig);
}
@@ -98,7 +98,7 @@ public class SessionJobReconciler
FlinkSessionJob cr,
FlinkSessionJobSpec sessionJobSpec,
FlinkSessionJobStatus status,
- Context ctx,
+ Context<?> ctx,
Configuration deployConfig,
Optional<String> savepoint,
boolean requireHaMetadata)
@@ -118,7 +118,7 @@ public class SessionJobReconciler
@Override
protected void cancelJob(
FlinkSessionJob resource,
- Context ctx,
+ Context<?> ctx,
UpgradeMode upgradeMode,
Configuration observeConfig)
throws Exception {
@@ -127,7 +127,7 @@ public class SessionJobReconciler
}
@Override
- public DeleteControl cleanupInternal(FlinkSessionJob sessionJob, Context
context) {
+ public DeleteControl cleanupInternal(FlinkSessionJob sessionJob,
Context<?> context) {
Optional<FlinkDeployment> flinkDepOptional =
context.getSecondaryResource(FlinkDeployment.class);
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 c4b32529..1a601446 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
@@ -99,7 +99,6 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
-import java.net.SocketTimeoutException;
import java.net.URI;
import java.util.Arrays;
import java.util.Collection;
@@ -202,10 +201,9 @@ public abstract class AbstractFlinkService implements
FlinkService {
socket.connect(socketAddress, 1000);
socket.close();
return true;
- } catch (SocketTimeoutException ste) {
} catch (IOException e) {
+ return false;
}
- return false;
}
@Override
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
index c96fec07..691bca94 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
@@ -24,8 +24,6 @@ import
org.apache.flink.kubernetes.operator.listener.FlinkResourceListener;
import io.fabric8.kubernetes.api.model.Event;
import io.fabric8.kubernetes.client.KubernetesClient;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.function.BiConsumer;
@@ -33,8 +31,6 @@ import java.util.function.BiConsumer;
/** Helper class for creating Kubernetes events for Flink resources. */
public class EventRecorder {
- private static final Logger LOG =
LoggerFactory.getLogger(EventRecorder.class);
-
private final KubernetesClient client;
private final BiConsumer<AbstractFlinkResource<?, ?>, Event> eventListener;
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 18fc9e07..496dbc1d 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
@@ -96,6 +96,7 @@ public class DefaultValidator implements
FlinkResourceValidator {
validateServiceAccount(spec.getServiceAccount()));
}
+ @SafeVarargs
private static Optional<String> firstPresent(Optional<String>... errOpts) {
for (Optional<String> opt : errOpts) {
if (opt.isPresent()) {
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 f3baba7d..6a94fe3c 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
@@ -20,7 +20,6 @@ package org.apache.flink.kubernetes.operator.observer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
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;
@@ -56,9 +55,7 @@ public class SavepointObserverTest {
@BeforeEach
public void before() {
var eventRecorder = new EventRecorder(kubernetesClient, (r, e) -> {});
- observer =
- new SavepointObserver<>(
- flinkService, configManager, new
TestingStatusRecorder<>(), eventRecorder);
+ observer = new SavepointObserver<>(flinkService, configManager,
eventRecorder);
}
@Test
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 19827276..3a72a520 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
@@ -22,7 +22,6 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
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.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.spec.JobState;
@@ -68,9 +67,7 @@ public class ApplicationObserverTest {
@BeforeEach
public void before() {
var eventRecorder = new EventRecorder(kubernetesClient, (r, e) -> {});
- observer =
- new ApplicationObserver(
- flinkService, configManager, new
TestingStatusRecorder<>(), eventRecorder);
+ observer = new ApplicationObserver(flinkService, configManager,
eventRecorder);
}
@Test
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 f970bb66..dec08ef8 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
@@ -63,9 +63,7 @@ public class SessionJobObserverTest {
var statusRecorder = new TestingStatusRecorder<FlinkSessionJob,
FlinkSessionJobStatus>();
flinkService = new TestingFlinkService();
FlinkServiceFactory flinkServiceFactory = new
TestingFlinkServiceFactory(flinkService);
- observer =
- new SessionJobObserver(
- flinkServiceFactory, configManager, statusRecorder,
eventRecorder);
+ observer = new SessionJobObserver(flinkServiceFactory, configManager,
eventRecorder);
reconciler =
new SessionJobReconciler(
kubernetesClient,
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 fbfcbdeb..e473d47f 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
@@ -54,8 +54,7 @@ public class StandaloneKubernetesJobManagerParameters extends
KubernetesJobManag
@Override
public Map<String, String> getSelectors() {
- return Collections.unmodifiableMap(
-
StandaloneKubernetesUtils.getJobManagerSelectors(getClusterId()));
+ return
StandaloneKubernetesUtils.getJobManagerSelectors(getClusterId());
}
@Override
diff --git
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/standalone/KubernetesStandaloneClusterDescriptor.java
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/standalone/KubernetesStandaloneClusterDescriptor.java
index cfed561a..1bed020b 100644
---
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/standalone/KubernetesStandaloneClusterDescriptor.java
+++
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/standalone/KubernetesStandaloneClusterDescriptor.java
@@ -94,8 +94,7 @@ public class KubernetesStandaloneClusterDescriptor extends
KubernetesClusterDesc
StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE,
StandaloneKubernetesConfigOptionsInternal.ClusterMode.SESSION);
- final ClusterClientProvider clusterClientProvider =
- deployClusterInternal(clusterSpecification);
+ var clusterClientProvider =
deployClusterInternal(clusterSpecification);
try (ClusterClient<String> clusterClient =
clusterClientProvider.getClusterClient()) {
LOG.info(
@@ -115,8 +114,7 @@ public class KubernetesStandaloneClusterDescriptor extends
KubernetesClusterDesc
StandaloneKubernetesConfigOptionsInternal.CLUSTER_MODE,
StandaloneKubernetesConfigOptionsInternal.ClusterMode.APPLICATION);
applicationConfiguration.applyToConfiguration(flinkConfig);
- final ClusterClientProvider clusterClientProvider =
- deployClusterInternal(clusterSpecification);
+ var clusterClientProvider =
deployClusterInternal(clusterSpecification);
try (ClusterClient<String> clusterClient =
clusterClientProvider.getClusterClient()) {
LOG.info(
diff --git
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/utils/StandaloneKubernetesUtils.java
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/utils/StandaloneKubernetesUtils.java
index ca801e4c..0a743435 100644
---
a/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/utils/StandaloneKubernetesUtils.java
+++
b/flink-kubernetes-standalone/src/main/java/org/apache/flink/kubernetes/operator/utils/StandaloneKubernetesUtils.java
@@ -38,7 +38,7 @@ public class StandaloneKubernetesUtils {
}
public static Map<String, String> getCommonLabels(String clusterId) {
- Map<String, String> commonLabels = new HashMap();
+ Map<String, String> commonLabels = new HashMap<>();
commonLabels.put(Constants.LABEL_TYPE_KEY, LABEL_TYPE_STANDALONE_TYPE);
commonLabels.put(Constants.LABEL_APP_KEY, clusterId);
return commonLabels;
@@ -55,10 +55,4 @@ public class StandaloneKubernetesUtils {
labels.put(Constants.LABEL_COMPONENT_KEY,
Constants.LABEL_COMPONENT_JOB_MANAGER);
return Collections.unmodifiableMap(labels);
}
-
- public static Map<String, String> getConfigMapLabels(String clusterId,
String type) {
- Map<String, String> labels = new HashMap(getCommonLabels(clusterId));
- labels.put("configmap-type", type);
- return Collections.unmodifiableMap(labels);
- }
}
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 832ce97a..80b5ab80 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
@@ -175,7 +175,7 @@ public class StandaloneKubernetesJobManagerFactoryTest
extends ParametersTestBas
}
@Test
- public void testAdditionalResourcesSize() throws IOException {
+ public void testAdditionalResourcesSize() {
final List<HasMetadata> resultAdditionalResources =
this.jmSpec.getAccompanyingResources();
assertEquals(3, resultAdditionalResources.size());
diff --git
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesTaskManagerParametersTest.java
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesTaskManagerParametersTest.java
index c794a630..bc71bd1c 100644
---
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesTaskManagerParametersTest.java
+++
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/parameters/StandaloneKubernetesTaskManagerParametersTest.java
@@ -99,9 +99,7 @@ public class StandaloneKubernetesTaskManagerParametersTest
extends ParametersTes
assertThrows(
IllegalArgumentException.class,
- () -> {
- kubernetesTaskManagerParameters.getReplicas();
- });
+ () -> kubernetesTaskManagerParameters.getReplicas());
}
@Test
@@ -134,10 +132,7 @@ public class StandaloneKubernetesTaskManagerParametersTest
extends ParametersTes
public void testInvalidRPCPort() {
flinkConfig.set(TaskManagerOptions.RPC_PORT, String.valueOf(0));
assertThrows(
- IllegalArgumentException.class,
- () -> {
- kubernetesTaskManagerParameters.getRPCPort();
- });
+ IllegalArgumentException.class, () ->
kubernetesTaskManagerParameters.getRPCPort());
}
@Test
diff --git
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/utils/TestUtils.java
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/utils/TestUtils.java
index 5a651930..6b777f34 100644
---
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/utils/TestUtils.java
+++
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/kubeclient/utils/TestUtils.java
@@ -53,7 +53,7 @@ public class TestUtils {
public static final double TASK_MANAGER_CPU = 4;
public static final double JOB_MANAGER_CPU = 2;
- public static final Map<String, String> generateTestStringStringMap(
+ public static Map<String, String> generateTestStringStringMap(
String keyPrefix, String valuePrefix, int entries) {
Map<String, String> map = new HashMap<>();
for (int i = 1; i <= entries; i++) {
diff --git
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/standalone/KubernetesStandaloneClusterDescriptorTest.java
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/standalone/KubernetesStandaloneClusterDescriptorTest.java
index 8aae921e..641ddfdc 100644
---
a/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/standalone/KubernetesStandaloneClusterDescriptorTest.java
+++
b/flink-kubernetes-standalone/src/test/java/org/apache/flink/kubernetes/operator/standalone/KubernetesStandaloneClusterDescriptorTest.java
@@ -19,8 +19,6 @@ package org.apache.flink.kubernetes.operator.standalone;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
-import org.apache.flink.client.program.ClusterClient;
-import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
@@ -76,8 +74,7 @@ public class KubernetesStandaloneClusterDescriptorTest {
flinkConfig.setString(TaskManagerOptions.RPC_PORT, String.valueOf(0));
flinkConfig.setString(RestOptions.BIND_PORT, String.valueOf(0));
- ClusterClientProvider clusterClientProvider =
- clusterDescriptor.deploySessionCluster(clusterSpecification);
+ var clusterClientProvider =
clusterDescriptor.deploySessionCluster(clusterSpecification);
List<Deployment> deployments =
kubernetesClient
@@ -112,7 +109,7 @@ public class KubernetesStandaloneClusterDescriptorTest {
jmDeployment.getSpec().getTemplate().getSpec().getContainers().stream()
.anyMatch(c -> c.getArgs().contains("jobmanager")));
- ClusterClient clusterClient = clusterClientProvider.getClusterClient();
+ var clusterClient = clusterClientProvider.getClusterClient();
String expectedWebUrl =
String.format(
@@ -131,7 +128,7 @@ public class KubernetesStandaloneClusterDescriptorTest {
flinkConfig.setString(TaskManagerOptions.RPC_PORT, String.valueOf(0));
flinkConfig.setString(RestOptions.BIND_PORT, String.valueOf(0));
- ClusterClientProvider clusterClientProvider =
+ var clusterClientProvider =
clusterDescriptor.deployApplicationCluster(
clusterSpecification,
ApplicationConfiguration.fromConfiguration(flinkConfig));
@@ -169,7 +166,7 @@ public class KubernetesStandaloneClusterDescriptorTest {
jmDeployment.getSpec().getTemplate().getSpec().getContainers().stream()
.anyMatch(c ->
c.getArgs().contains("standalone-job")));
- ClusterClient clusterClient = clusterClientProvider.getClusterClient();
+ var clusterClient = clusterClientProvider.getClusterClient();
String expectedWebUrl =
String.format(
diff --git
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandler.java
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandler.java
index 2a877c91..64ca21b7 100644
---
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandler.java
+++
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/AdmissionHandler.java
@@ -72,7 +72,7 @@ public class AdmissionHandler extends
SimpleChannelInboundHandler<HttpRequest> {
public AdmissionHandler(Validator<HasMetadata> validator,
Mutator<HasMetadata> mutator) {
this.validatingController = new AdmissionController<>(validator);
- this.mutatorController = new AdmissionController<>(new
DefaultRequestMutator(mutator));
+ this.mutatorController = new AdmissionController<>(new
DefaultRequestMutator<>(mutator));
}
@Override
diff --git
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
index df2f9e32..9a7c03db 100644
---
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
+++
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
@@ -107,7 +107,7 @@ public class FlinkOperatorWebhook {
AdmissionHandler admissionHandler) throws Exception {
SslContext sslContext = createSslContext();
- return new ChannelInitializer<SocketChannel>() {
+ return new ChannelInitializer<>() {
@Override
protected void initChannel(SocketChannel ch) {
diff --git
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManager.java
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManager.java
index 2915414d..1b3ae4d9 100644
---
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManager.java
+++
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManager.java
@@ -108,13 +108,11 @@ public class InformerManager {
if (flinkDepInformers != null) {
synchronized (this) {
if (flinkDepInformers != null) {
- flinkDepInformers
- .entrySet()
- .forEach(
- entry -> {
- LOG.info("Stopping informer in {})",
entry.getKey());
- entry.getValue().stop();
- });
+ flinkDepInformers.forEach(
+ (key, value) -> {
+ LOG.info("Stopping informer in {})", key);
+ value.stop();
+ });
}
flinkDepInformers = null;
}
diff --git
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/mutator/DefaultRequestMutator.java
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/mutator/DefaultRequestMutator.java
index bcff856a..5cba82bc 100644
---
a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/mutator/DefaultRequestMutator.java
+++
b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/mutator/DefaultRequestMutator.java
@@ -61,12 +61,11 @@ public class DefaultRequestMutator<T extends
KubernetesResource> implements Requ
Operation operation =
Operation.valueOf(admissionRequest.getOperation());
KubernetesResource originalResource =
AdmissionUtils.getTargetResource(admissionRequest, operation);
- KubernetesResource clonedResource =
- (KubernetesResource) this.cloner.clone((T) originalResource);
+ T clonedResource = this.cloner.clone((T) originalResource);
AdmissionResponse admissionResponse;
try {
- T mutatedResource = this.mutator.mutate((T) clonedResource,
operation);
+ T mutatedResource = this.mutator.mutate(clonedResource, operation);
admissionResponse =
admissionResponseFromMutation(originalResource, mutatedResource);
} catch (NotAllowedException e) {
admissionResponse =
AdmissionUtils.notAllowedExceptionToAdmissionResponse(e);
diff --git
a/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManagerTest.java
b/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManagerTest.java
index d258d6cb..bbb5971e 100644
---
a/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManagerTest.java
+++
b/flink-kubernetes-webhook/src/test/java/org/apache/flink/kubernetes/operator/admission/informer/InformerManagerTest.java
@@ -62,7 +62,7 @@ public class InformerManagerTest {
Configuration config =
Configuration.fromMap(Map.of(OPERATOR_WATCHED_NAMESPACES.key(), "ns1"));
FlinkConfigManager configManager =
- new FlinkConfigManager(config, ns ->
informerManager.setNamespaces(ns));
+ new FlinkConfigManager(config, informerManager::setNamespaces);
informerManager.setNamespaces(
configManager.getOperatorConfiguration().getWatchedNamespaces());
Assertions.assertNotNull(informerManager.getFlinkDepInformer("ns1"));
diff --git a/tools/maven/checkstyle.xml b/tools/maven/checkstyle.xml
index 2048fd18..759fea16 100644
--- a/tools/maven/checkstyle.xml
+++ b/tools/maven/checkstyle.xml
@@ -169,6 +169,11 @@ This file is based on the checkstyle file of Apache Beam.
<property name="illegalPattern" value="true"/>
<property name="message" value="Use
com.fasterxml.jackson instead of jettison."/>
</module>
+ <module name="Regexp">
+ <property name="format" value="org\.junit\.Assert\."/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use
org.junit.jupiter.api.Assertions."/>
+ </module>
<!-- Enforce Java-style array declarations -->
<module name="ArrayTypeStyle"/>
