Updated to fix Kubernetes scale issue (#3607)

Author: nicknezis

heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/V1Controller.java

@@ -66,6 +66,7 @@
 import io.kubernetes.client.openapi.models.V1Toleration;
 import io.kubernetes.client.openapi.models.V1Volume;
 import io.kubernetes.client.openapi.models.V1VolumeMount;
+import io.kubernetes.client.util.PatchUtils;
 
 import okhttp3.Response;
 
@@ -159,11 +160,8 @@ boolean restart(int shardId) {
     final int currentContainerCount = statefulSet.getSpec().getReplicas();
     final int newContainerCount = currentContainerCount + containersToAdd.size();
 
-    final V1StatefulSetSpec newSpec = new V1StatefulSetSpec();
-    newSpec.setReplicas(newContainerCount);
-
     try {
-      doPatch(newSpec);
+      patchStatefulsetReplicas(newContainerCount);
     } catch (ApiException ae) {
       throw new TopologyRuntimeManagementException(
           ae.getMessage() + "\ndetails\n" + ae.getResponseBody());
@@ -184,28 +182,37 @@ public void removeContainers(Set<PackingPlan.ContainerPlan> containersToRemove)
     final int currentContainerCount = statefulSet.getSpec().getReplicas();
     final int newContainerCount = currentContainerCount - containersToRemove.size();
 
-    final V1StatefulSetSpec newSpec = new V1StatefulSetSpec();
-    newSpec.setReplicas(newContainerCount);
-
     try {
-      doPatch(newSpec);
+      patchStatefulsetReplicas(newContainerCount);
     } catch (ApiException e) {
       throw new TopologyRuntimeManagementException(
           e.getMessage() + "\ndetails\n" + e.getResponseBody());
     }
   }
 
-  private void doPatch(V1StatefulSetSpec patchedSpec) throws ApiException {
+  private void patchStatefulsetReplicas(int replicas) throws ApiException {
     final String body =
             String.format(JSON_PATCH_STATEFUL_SET_REPLICAS_FORMAT,
-                    patchedSpec.getReplicas().toString());
+                    replicas);
     final V1Patch patch = new V1Patch(body);
-    appsClient.patchNamespacedStatefulSet(getTopologyName(),
-            getNamespace(), patch, null, null, null, null);
+
+    PatchUtils.patch(V1StatefulSet.class,
+        () ->
+          appsClient.patchNamespacedStatefulSetCall(
+            getTopologyName(),
+            getNamespace(),
+            patch,
+            null,
+            null,
+            null,
+            null,
+            null),
+        V1Patch.PATCH_FORMAT_JSON_PATCH,
+        appsClient.getApiClient());
   }
 
   private static final String JSON_PATCH_STATEFUL_SET_REPLICAS_FORMAT =
-      "{\"op\":\"replace\",\"path\":\"/spec/replicas\",\"value\":%s}";
+          "[{\"op\":\"replace\",\"path\":\"/spec/replicas\",\"value\":%d}]";
 
   V1StatefulSet getStatefulSet() throws ApiException {
     return appsClient.readNamespacedStatefulSet(getTopologyName(), getNamespace(),