Updating client side ProcessFunction and unit tests.

Author: schongloo

flink-kubernetes-operator-bluegreen-client/pom.xml

@@ -24,18 +24,25 @@ under the License.
     <parent>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-kubernetes-operator-parent</artifactId>
-        <version>1.14-SNAPSHOT</version>
+        <version>1.13-apple-SNAPSHOT</version>
         <relativePath>..</relativePath>
     </parent>
 
     <artifactId>flink-kubernetes-operator-bluegreen-client</artifactId>
     <name>Flink Kubernetes BlueGreen Deployment Client</name>
 
+    <properties>
+        <surefire.module.config>
+            <!-- required by Kryo/chill ArraysAsListSerializer in test harness -->
+            --add-opens=java.base/java.util=ALL-UNNAMED
+        </surefire.module.config>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-kubernetes-operator-api</artifactId>
-            <version>1.14-SNAPSHOT</version>
+            <version>1.13-apple-SNAPSHOT</version>
             <scope>compile</scope>
         </dependency>
 
@@ -70,23 +77,23 @@ under the License.
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-test-utils</artifactId>
+            <artifactId>flink-runtime</artifactId>
             <version>${flink.version}</version>
             <scope>test</scope>
         </dependency>
 
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java</artifactId>
+            <artifactId>flink-test-utils</artifactId>
             <version>${flink.version}</version>
-            <type>test-jar</type>
             <scope>test</scope>
         </dependency>
 
         <dependency>
-            <groupId>io.fabric8</groupId>
-            <artifactId>kubernetes-httpclient-okhttp</artifactId>
-            <version>${fabric8.version}</version>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java</artifactId>
+            <version>${flink.version}</version>
+            <type>test-jar</type>
             <scope>test</scope>
         </dependency>
     </dependencies>

flink-kubernetes-operator-bluegreen-client/src/main/java/org/apache/flink/kubernetes/operator/bluegreen/client/GateProcessFunction.java

@@ -17,6 +17,8 @@
 
 package org.apache.flink.kubernetes.operator.bluegreen.client;
 
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDeploymentType;
 import org.apache.flink.kubernetes.operator.api.bluegreen.GateContext;
@@ -25,9 +27,6 @@
 import org.apache.flink.streaming.api.functions.ProcessFunction;
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.Preconditions;
-
-import io.fabric8.kubernetes.api.model.ConfigMap;
-import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,8 +48,8 @@ abstract class GateProcessFunction<I> extends ProcessFunction<I, I> implements S
     // TODO: make this configurable? This cannot be a constant
     protected final int subtaskIndexGuide = 1;
 
-    protected GateKubernetesService gateKubernetesService;
-    protected Boolean clearToTeardown = false;
+    private GateKubernetesService gateKubernetesService;
+    private Boolean clearToTeardown = false;
     protected GateContext baseContext;
     private String namespace;
     private String configMapName;
@@ -60,8 +59,7 @@ abstract class GateProcessFunction<I> extends ProcessFunction<I, I> implements S
     public GateProcessFunction(
             BlueGreenDeploymentType blueGreenDeploymentType,
             String namespace,
-            String configMapName,
-            GateKubernetesService gateKubernetesService) {
+            String configMapName) {
         Preconditions.checkArgument(
                 blueGreenDeploymentType == BlueGreenDeploymentType.BLUE
                         || blueGreenDeploymentType == BlueGreenDeploymentType.GREEN,
@@ -70,19 +68,12 @@ public GateProcessFunction(
         this.blueGreenDeploymentType = blueGreenDeploymentType;
         this.namespace = namespace;
         this.configMapName = configMapName;
-        this.gateKubernetesService = gateKubernetesService;
     }
 
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
 
-        // Only create service if not injected
-        if (gateKubernetesService == null) {
-            this.gateKubernetesService = new GateKubernetesService(namespace, configMapName);
-        }
-
-        // Always set up informers, whether service was injected or created
         setKubernetesEnvironment();
         processConfigMap(gateKubernetesService.parseConfigMap());
     }
@@ -113,6 +104,8 @@ public void processElement(I value, ProcessFunction<I, I>.Context ctx, Collector
     }
 
     private void setKubernetesEnvironment() {
+        this.gateKubernetesService = new GateKubernetesService(namespace, configMapName);
+
         logInfo("Preparing Informers...");
         var resourceEventHandler =
                 new ResourceEventHandler<ConfigMap>() {
@@ -167,26 +160,18 @@ private void processConfigMap(ConfigMap configMap) {
         onContextUpdate(baseContext, filteredData);
     }
 
-    protected void notifyClearToTeardown() {
+    protected final void notifyClearToTeardown() {
         // Notify only once
         if (!clearToTeardown && getRuntimeContext().getIndexOfThisSubtask() == subtaskIndexGuide) {
             logInfo("Setting " + CLEAR_TO_TEARDOWN);
-            performConfigMapUpdate(
+            gateKubernetesService.updateConfigMapEntries(
                     Map.of(TRANSITION_STAGE.getLabel(), CLEAR_TO_TEARDOWN.toString()));
             logInfo(CLEAR_TO_TEARDOWN + " set!");
             this.clearToTeardown = true;
         }
     }
 
-    /**
-     * Template method for updating ConfigMap entries. Override in tests to avoid actual K8s
-     * updates.
-     */
-    protected void performConfigMapUpdate(Map<String, String> updates) {
-        gateKubernetesService.updateConfigMapEntries(updates);
-    }
-
-    protected void updateConfigMapCustomEntries(Map<String, String> customEntries)
+    protected final void updateConfigMapCustomEntries(Map<String, String> customEntries)
             throws IllegalAccessException {
         // Validating only "custom" entries/keys can be updated
         var keysToUpdate = customEntries.keySet();
@@ -203,7 +188,7 @@ protected void updateConfigMapCustomEntries(Map<String, String> customEntries)
             throw new IllegalAccessException(error);
         }
         logInfo("Updating custom entries: " + customEntries);
-        performConfigMapUpdate(customEntries);
+        gateKubernetesService.updateConfigMapEntries(customEntries);
     }
 
     // Temporary "utility" function for development

flink-kubernetes-operator-bluegreen-client/src/main/java/org/apache/flink/kubernetes/operator/bluegreen/client/WatermarkGateContext.java

@@ -17,13 +17,13 @@
 
 package org.apache.flink.kubernetes.operator.bluegreen.client;
 
-import org.apache.flink.kubernetes.operator.api.bluegreen.GateContext;
-
 import lombok.Getter;
 import lombok.Setter;
+import org.apache.flink.kubernetes.operator.api.bluegreen.GateContext;
 
 import java.io.Serializable;
 import java.util.Map;
+import java.util.Objects;
 
 /** Watermark based functionality of the GateContext. */
 @Getter
@@ -52,7 +52,7 @@ public boolean equals(Object o) {
             WatermarkGateContext wm = (WatermarkGateContext) o;
             return this.baseContext.equals(wm.getBaseContext())
                     && this.watermarkGateStage.equals(wm.getWatermarkGateStage())
-                    && this.watermarkToggleValue.equals(wm.getWatermarkToggleValue());
+                    && Objects.equals(this.watermarkToggleValue, wm.getWatermarkToggleValue());
         }
         return false;
     }

flink-kubernetes-operator-bluegreen-client/src/main/java/org/apache/flink/kubernetes/operator/bluegreen/client/WatermarkGateProcessFunction.java

@@ -19,7 +19,6 @@
 
 import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDeploymentType;
 import org.apache.flink.kubernetes.operator.api.bluegreen.GateContext;
-import org.apache.flink.kubernetes.operator.api.bluegreen.GateKubernetesService;
 import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionStage;
 import org.apache.flink.streaming.api.functions.ProcessFunction;
 import org.apache.flink.util.Collector;
@@ -35,25 +34,16 @@ public class WatermarkGateProcessFunction<I> extends GateProcessFunction<I>
 
     private final Function<I, Long> watermarkExtractor;
 
-    protected WatermarkGateContext currentWatermarkGateContext;
+    private WatermarkGateContext currentWatermarkGateContext;
 
-    protected Boolean waitingForWatermark = false;
+    private boolean waitingForWatermark = false;
 
     WatermarkGateProcessFunction(
             BlueGreenDeploymentType blueGreenDeploymentType,
             String namespace,
             String configMapName,
             Function<I, Long> watermarkExtractor) {
-        this(blueGreenDeploymentType, namespace, configMapName, watermarkExtractor, null);
-    }
-
-    WatermarkGateProcessFunction(
-            BlueGreenDeploymentType blueGreenDeploymentType,
-            String namespace,
-            String configMapName,
-            Function<I, Long> watermarkExtractor,
-            GateKubernetesService gateKubernetesService) {
-        super(blueGreenDeploymentType, namespace, configMapName, gateKubernetesService);
+        super(blueGreenDeploymentType, namespace, configMapName);
 
         Preconditions.checkNotNull(watermarkExtractor);
 
@@ -147,7 +137,7 @@ private long getWatermarkBoundary(ProcessFunction<I, I>.Context ctx) {
                 : ctx.timerService().currentProcessingTime();
     }
 
-    private void updateWatermarkInConfigMap(ProcessFunction<I, I>.Context ctx)
+    protected void updateWatermarkInConfigMap(ProcessFunction<I, I>.Context ctx)
             throws IllegalAccessException {
         var curWmCtx = currentWatermarkGateContext;
 
@@ -173,7 +163,7 @@ private void updateWatermarkInConfigMap(ProcessFunction<I, I>.Context ctx)
         }
     }
 
-    private void notifyWaitingForWatermark() throws IllegalAccessException {
+    protected void notifyWaitingForWatermark() throws IllegalAccessException {
         if (!waitingForWatermark
                 && getRuntimeContext().getIndexOfThisSubtask() == subtaskIndexGuide) {
             logInfo("Setting " + WatermarkGateStage.WAITING_FOR_WATERMARK);