[fix][fn] Fix k8s merge runtime opts bug (apache#19481)

Fixes: apache#19478

### Motivation

See issue for additional context. Essentially, we are doing a shallow clone when we needed a deep clone. The consequence is leaked labels, annotations, and tolerations.

### Modifications

* Add a `deepClone` method to the `BasicKubernetesManifestCustomizer.RuntimeOpts` method. Note that this method is not technically a deep clone for the k8s objects. However, based on the way we "merge" these objects, it is sufficient to copy references to the objects.

### Verifying this change

Added a test that fails before the change and passes afterwards.

### Documentation

- [x] `doc-not-needed`

This is an internal bug fix. No docs needed.

### Matching PR in forked repository

PR in forked repository: michaeljmarshall#27

pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/BasicKubernetesManifestCustomizer.java

@@ -62,7 +62,7 @@ public class BasicKubernetesManifestCustomizer implements KubernetesManifestCust
     @Setter
     @NoArgsConstructor
     @AllArgsConstructor
-    @Builder(toBuilder = true)
+    @Builder()
     public static class RuntimeOpts {
         private String jobNamespace;
         private String jobName;
@@ -71,6 +71,21 @@ public static class RuntimeOpts {
         private Map<String, String> nodeSelectorLabels;
         private V1ResourceRequirements resourceRequirements;
         private List<V1Toleration> tolerations;
+
+        /**
+         * A clone where the maps and lists are properly cloned. The k8s resources themselves are shallow clones.
+         */
+        public RuntimeOpts partialDeepClone() {
+            return new RuntimeOpts(
+                    jobNamespace,
+                    jobName,
+                    extraLabels != null ? new HashMap<>(extraLabels) : null,
+                    extraAnnotations != null ? new HashMap<>(extraAnnotations) : null,
+                    nodeSelectorLabels != null ? new HashMap<>(nodeSelectorLabels) : null,
+                    resourceRequirements,
+                    tolerations != null ? new ArrayList<>(tolerations) : null
+            );
+        }
     }
 
     @Getter
@@ -82,7 +97,7 @@ public void initialize(Map<String, Object> config) {
             RuntimeOpts opts =
                     ObjectMapperFactory.getMapper().getObjectMapper().convertValue(config, RuntimeOpts.class);
             if (opts != null) {
-                runtimeOpts = opts.toBuilder().build();
+                runtimeOpts = opts;
             }
         } else {
             log.warn("initialize with null config");
@@ -176,7 +191,7 @@ private V1ObjectMeta updateMeta(RuntimeOpts opts, V1ObjectMeta meta) {
     }
 
     public static RuntimeOpts mergeRuntimeOpts(RuntimeOpts oriOpts, RuntimeOpts newOpts) {
-        RuntimeOpts mergedOpts = oriOpts.toBuilder().build();
+        RuntimeOpts mergedOpts = oriOpts.partialDeepClone();
         if (mergedOpts.getExtraLabels() == null) {
             mergedOpts.setExtraLabels(new HashMap<>());
         }

pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/kubernetes/BasicKubernetesManifestCustomizerTest.java

@@ -24,8 +24,10 @@
 import io.kubernetes.client.openapi.models.V1Toleration;
 import org.testng.annotations.Test;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import static org.testng.Assert.assertEquals;
@@ -93,4 +95,55 @@ public void TestMergeRuntimeOpts() {
         assertEquals(mergedOpts.getResourceRequirements().getLimits().get("cpu").getNumber().intValue(), 20);
         assertEquals(mergedOpts.getResourceRequirements().getLimits().get("memory").getNumber().intValue(), 10240);
     }
+
+    // Note: this test creates many new objects to ensure that the tests guarantees objects are not mutated
+    // unexpectedly.
+    @Test
+    public void testMergeRuntimeOptsDoesNotModifyArguments() {
+        BasicKubernetesManifestCustomizer.RuntimeOpts opts1 = new BasicKubernetesManifestCustomizer.RuntimeOpts(
+                "namespace1", "job1", new HashMap<>(), new HashMap<>(), new HashMap<>(), new V1ResourceRequirements(),
+                new ArrayList<>());
+
+        HashMap<String, String> testMap = new HashMap<>();
+        testMap.put("testKey", "testValue");
+
+        List<V1Toleration> testList = new ArrayList<>();
+        testList.add(new V1Toleration());
+
+        V1ResourceRequirements requirements = new V1ResourceRequirements();
+        requirements.setLimits(new HashMap<>());
+        BasicKubernetesManifestCustomizer.RuntimeOpts opts2 = new BasicKubernetesManifestCustomizer.RuntimeOpts(
+                "namespace2", "job2", testMap, testMap, testMap,requirements, testList);
+
+        // Merge the runtime opts
+        BasicKubernetesManifestCustomizer.RuntimeOpts result =
+                BasicKubernetesManifestCustomizer.mergeRuntimeOpts(opts1, opts2);
+
+        // Assert opts1 is same
+        assertEquals("namespace1", opts1.getJobNamespace());
+        assertEquals("job1", opts1.getJobName());
+        assertEquals(new HashMap<>(), opts1.getNodeSelectorLabels());
+        assertEquals(new HashMap<>(), opts1.getExtraAnnotations());
+        assertEquals(new HashMap<>(), opts1.getExtraLabels());
+        assertEquals(new ArrayList<>(), opts1.getTolerations());
+        assertEquals(new V1ResourceRequirements(), opts1.getResourceRequirements());
+
+        // Assert opts2 is same
+        HashMap<String, String> expectedTestMap = new HashMap<>();
+        expectedTestMap.put("testKey", "testValue");
+
+        List<V1Toleration> expectedTestList = new ArrayList<>();
+        expectedTestList.add(new V1Toleration());
+
+        V1ResourceRequirements expectedRequirements = new V1ResourceRequirements();
+        expectedRequirements.setLimits(new HashMap<>());
+
+        assertEquals("namespace2", opts2.getJobNamespace());
+        assertEquals("job2", opts2.getJobName());
+        assertEquals(expectedTestMap, opts2.getNodeSelectorLabels());
+        assertEquals(expectedTestMap, opts2.getExtraAnnotations());
+        assertEquals(expectedTestMap, opts2.getExtraLabels());
+        assertEquals(expectedTestList, opts2.getTolerations());
+        assertEquals(expectedRequirements, opts2.getResourceRequirements());
+    }
 }