ingest: The IngestDocument copy constructor should make a deep copy instead of shallow copy

Closes #16246

core/src/main/java/org/elasticsearch/ingest/core/IngestDocument.java

@@ -81,7 +81,7 @@ public final class IngestDocument {
      * Copy constructor that creates a new {@link IngestDocument} which has exactly the same properties as the one provided as argument
      */
     public IngestDocument(IngestDocument other) {
-        this(new HashMap<>(other.sourceAndMetadata), new HashMap<>(other.ingestMetadata));
+        this(deepCopyMap(other.sourceAndMetadata), deepCopyMap(other.ingestMetadata));
     }
 
     /**
@@ -470,6 +470,35 @@ public final class IngestDocument {
         return this.sourceAndMetadata;
     }
 
+    @SuppressWarnings("unchecked")
+    private static <K, V> Map<K, V> deepCopyMap(Map<K, V> source) {
+        return (Map<K, V>) deepCopy(source);
+    }
+
+    private static Object deepCopy(Object value) {
+        if (value instanceof Map) {
+            Map<?, ?> mapValue = (Map<?, ?>) value;
+            Map<Object, Object> copy = new HashMap<>(mapValue.size());
+            for (Map.Entry<?, ?> entry : mapValue.entrySet()) {
+                copy.put(entry.getKey(), deepCopy(entry.getValue()));
+            }
+            return copy;
+        } else if (value instanceof List) {
+            List<?> listValue = (List<?>) value;
+            List<Object> copy = new ArrayList<>(listValue.size());
+            for (Object itemValue : listValue) {
+                copy.add(deepCopy(itemValue));
+            }
+            return copy;
+        } else if (value == null || value instanceof String || value instanceof Integer ||
+            value instanceof Long || value instanceof Float ||
+            value instanceof Double || value instanceof Boolean) {
+            return value;
+        } else {
+            throw new IllegalArgumentException("unexpected value type [" + value.getClass() + "]");
+        }
+    }
+
     @Override
     public boolean equals(Object obj) {
         if (obj == this) { return true; }

core/src/test/java/org/elasticsearch/ingest/core/IngestDocumentTests.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.ingest.core;
 
 import org.elasticsearch.ingest.RandomDocumentPicks;
-import org.elasticsearch.ingest.core.IngestDocument;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
 
@@ -970,7 +969,31 @@ public class IngestDocumentTests extends ESTestCase {
     public void testCopyConstructor() {
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
         IngestDocument copy = new IngestDocument(ingestDocument);
-        assertThat(ingestDocument.getSourceAndMetadata(), not(sameInstance(copy.getSourceAndMetadata())));
-        assertThat(ingestDocument.getSourceAndMetadata(), equalTo(copy.getSourceAndMetadata()));
+        recursiveEqualsButNotSameCheck(ingestDocument.getSourceAndMetadata(), copy.getSourceAndMetadata());
+    }
+
+    private void recursiveEqualsButNotSameCheck(Object a, Object b) {
+        assertThat(a, not(sameInstance(b)));
+        assertThat(a, equalTo(b));
+        if (a instanceof Map) {
+            Map<?, ?> mapA = (Map<?, ?>) a;
+            Map<?, ?> mapB = (Map<?, ?>) b;
+            for (Map.Entry<?, ?> entry : mapA.entrySet()) {
+                if (entry.getValue() instanceof List || entry.getValue() instanceof Map) {
+                    recursiveEqualsButNotSameCheck(entry.getValue(), mapB.get(entry.getKey()));
+                }
+            }
+        } else if (a instanceof List) {
+            List<?> listA = (List<?>) a;
+            List<?> listB = (List<?>) b;
+            for (int i = 0; i < listA.size(); i++) {
+                Object value = listA.get(i);
+                if (value instanceof List || value instanceof Map) {
+                    recursiveEqualsButNotSameCheck(value, listB.get(i));
+                }
+            }
+        }
+
     }
+
 }