More PR comments addressed

Author: schongloo

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/utils/SpecUtils.java

@@ -99,7 +99,7 @@ public static String writeSpecWithMeta(
         }
     }
 
-    public static String serializeObject(Object object, String wrapperKey) {
+    public static String writeSpecAsJSON(Object object, String wrapperKey) {
         ObjectNode wrapper = objectMapper.createObjectNode();
         wrapper.set(wrapperKey, objectMapper.valueToTree(checkNotNull(object)));
 
@@ -111,8 +111,7 @@ public static String serializeObject(Object object, String wrapperKey) {
         }
     }
 
-    public static <T> T deserializeObject(String serialized, String wrapperKey, Class<T> valueType)
-            throws JsonProcessingException {
+    public static <T> T readSpecFromJSON(String serialized, String wrapperKey, Class<T> valueType) {
         try {
             ObjectNode wrapper = (ObjectNode) objectMapper.readTree(serialized);
             return objectMapper.treeToValue(wrapper.get(wrapperKey), valueType);

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java

@@ -178,7 +178,7 @@ private UpdateControl<FlinkBlueGreenDeployment> monitorTransition(
             // this means the spec was changed during transition,
             //  ignore the new change, revert the spec and log as warning
             bgDeployment.setSpec(
-                    SpecUtils.deserializeObject(
+                    SpecUtils.readSpecFromJSON(
                             deploymentStatus.getLastReconciledSpec(),
                             "spec",
                             FlinkBlueGreenDeploymentSpec.class));
@@ -407,7 +407,7 @@ private static void setLastReconciledSpec(
             FlinkBlueGreenDeployment bgDeployment,
             FlinkBlueGreenDeploymentStatus deploymentStatus) {
         deploymentStatus.setLastReconciledSpec(
-                SpecUtils.serializeObject(bgDeployment.getSpec(), "spec"));
+                SpecUtils.writeSpecAsJSON(bgDeployment.getSpec(), "spec"));
         deploymentStatus.setLastReconciledTimestamp(System.currentTimeMillis());
     }
 
@@ -467,7 +467,7 @@ private boolean hasSpecChanged(
             FlinkBlueGreenDeploymentSpec newSpec, FlinkBlueGreenDeploymentStatus deploymentStatus) {
 
         String lastReconciledSpec = deploymentStatus.getLastReconciledSpec();
-        String newSpecSerialized = SpecUtils.serializeObject(newSpec, "spec");
+        String newSpecSerialized = SpecUtils.writeSpecAsJSON(newSpec, "spec");
 
         return !lastReconciledSpec.equals(newSpecSerialized);
     }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentUtils.java

@@ -52,8 +52,8 @@ public static <T> T adjustNameReferences(
             String wrapperKey,
             Class<T> valueType)
             throws JsonProcessingException {
-        String serializedSpec = SpecUtils.serializeObject(spec, wrapperKey);
+        String serializedSpec = SpecUtils.writeSpecAsJSON(spec, wrapperKey);
         String replacedSerializedSpec = serializedSpec.replace(deploymentName, childDeploymentName);
-        return SpecUtils.deserializeObject(replacedSerializedSpec, wrapperKey, valueType);
+        return SpecUtils.readSpecFromJSON(replacedSerializedSpec, wrapperKey, valueType);
     }
 }

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java

@@ -269,8 +269,8 @@ public void verifySpecChangeDuringTransition(FlinkVersion flinkVersion) throws E
 
         // The spec should have been reverted
         assertEquals(
-                SpecUtils.serializeObject(originalSpec, "spec"),
-                SpecUtils.serializeObject(rs.deployment.getSpec(), "spec"));
+                SpecUtils.writeSpecAsJSON(originalSpec, "spec"),
+                SpecUtils.writeSpecAsJSON(rs.deployment.getSpec(), "spec"));
     }
 
     @ParameterizedTest
@@ -414,7 +414,7 @@ private void assertDeploymentDeleted(
         assertEquals(expectedDeletionDelay, deletionDelay);
         assertTrue(rs.reconciledStatus.getDeploymentReadyTimestamp() > 0);
         assertEquals(
-                SpecUtils.serializeObject(bgSpecBefore, "spec"),
+                SpecUtils.writeSpecAsJSON(bgSpecBefore, "spec"),
                 rs.reconciledStatus.getLastReconciledSpec());
 
         // A reconciliation before the deletion delay has expired should result in no-op
@@ -482,7 +482,7 @@ private void testTransitionToGreen(
         assertTrue(minReconciliationTs < rs.reconciledStatus.getLastReconciledTimestamp());
         assertNotNull(rs.reconciledStatus.getLastReconciledSpec());
         assertEquals(
-                SpecUtils.serializeObject(bgUpdatedSpec, "spec"),
+                SpecUtils.writeSpecAsJSON(bgUpdatedSpec, "spec"),
                 rs.reconciledStatus.getLastReconciledSpec());
         assertEquals(
                 FlinkBlueGreenDeploymentState.ACTIVE_GREEN,