Apply review comments

Author: eyalkoren

modules/ingest-otel/src/main/java/org/elasticsearch/ingest/otel/NormalizeToOTelProcessor.java

@@ -114,9 +114,8 @@ public IngestDocument execute(IngestDocument document) {
             if (keepKey.equals("@timestamp")) {
                 continue;
             }
-            Object value = source.remove(keepKey);
-            if (value != null) {
-                newAttributes.put(keepKey, value);
+            if (source.containsKey(keepKey)) {
+                newAttributes.put(keepKey, source.remove(keepKey));
             }
         }
 
@@ -147,8 +146,8 @@ public IngestDocument execute(IngestDocument document) {
         }
 
         // Flatten attributes
-        source.replace(ATTRIBUTES_KEY, Maps.flatten(newAttributes, false, false));
-        newResource.replace(ATTRIBUTES_KEY, Maps.flatten(newResourceAttributes, false, false));
+        source.put(ATTRIBUTES_KEY, Maps.flatten(newAttributes, false, false));
+        newResource.put(ATTRIBUTES_KEY, Maps.flatten(newResourceAttributes, false, false));
 
         return document;
     }
@@ -223,9 +222,12 @@ static boolean isOTelDocument(Map<String, Object> source) {
      */
     static void renameSpecialKeys(IngestDocument document) {
         RENAME_KEYS.forEach((nonOtelName, otelName) -> {
+            boolean fieldExists = false;
+            Object value = null;
             // first look assuming dot notation for nested fields
-            Object value = document.getFieldValue(nonOtelName, Object.class, true);
-            if (value != null) {
+            if (document.hasField(nonOtelName)) {
+                fieldExists = true;
+                value = document.getFieldValue(nonOtelName, Object.class, true);
                 document.removeField(nonOtelName);
                 // recursively remove empty parent fields
                 int lastDot = nonOtelName.lastIndexOf('.');
@@ -243,9 +245,13 @@ static void renameSpecialKeys(IngestDocument document) {
                 }
             } else if (nonOtelName.contains(".")) {
                 // look for dotted field names
-                value = document.getSource().remove(nonOtelName);
+                Map<String, Object> source = document.getSource();
+                if (source.containsKey(nonOtelName)) {
+                    fieldExists = true;
+                    value = source.remove(nonOtelName);
+                }
             }
-            if (value != null) {
+            if (fieldExists) {
                 document.setFieldValue(otelName, value);
             }
         });

modules/ingest-otel/src/test/java/org/elasticsearch/ingest/otel/NormalizeToOTelProcessorTests.java

@@ -17,7 +17,6 @@
 import java.util.Map;
 
 import static java.util.Map.entry;
-import static org.hamcrest.Matchers.sameInstance;
 
 public class NormalizeToOTelProcessorTests extends ESTestCase {
 
@@ -121,8 +120,10 @@ public void testExecute_validOTelDocument() {
             entry("key1", "value1")
         );
         IngestDocument document = new IngestDocument("index", "id", 1, null, null, source);
+        Map<String, Object> shallowCopy = new HashMap<>(source);
         processor.execute(document);
-        assertThat(source, sameInstance(document.getSource()));
+        // verify that top level keys are not moved when processing a valid OTel document
+        assertEquals(shallowCopy, document.getSource());
     }
 
     public void testExecute_nonOTelDocument() {
@@ -325,6 +326,35 @@ public void testExecute_moveToAttributeMaps() {
         assertFalse(source.containsKey("service"));
     }
 
+    public void testKeepNullValues() {
+        Map<String, Object> source = new HashMap<>();
+        Map<String, Object> span = new HashMap<>();
+        span.put("id", null);
+        source.put("span", span);
+        source.put("log.level", null);
+        source.put("trace_id", null);
+        source.put("foo", null);
+        source.put("agent.name", null);
+        IngestDocument document = new IngestDocument("index", "id", 1, null, null, source);
+
+        processor.execute(document);
+
+        assertFalse(source.containsKey("span"));
+        assertTrue(source.containsKey("span_id"));
+        assertNull(source.get("span_id"));
+        assertFalse(source.containsKey("log"));
+        assertTrue(source.containsKey("severity_text"));
+        assertNull(source.get("severity_text"));
+        assertFalse(source.containsKey("trace_id"));
+        Map<String, Object> expectedAttributes = new HashMap<>();
+        expectedAttributes.put("foo", null);
+        expectedAttributes.put("trace_id", null);
+        assertEquals(expectedAttributes, get(source, "attributes"));
+        Map<String, Object> expectedResourceAttributes = new HashMap<>();
+        expectedResourceAttributes.put("agent.name", null);
+        assertEquals(expectedResourceAttributes, get(get(source, "resource"), "attributes"));
+    }
+
     public void testExecute_deepFlattening() {
         Map<String, Object> source = new HashMap<>();
         Map<String, Object> nestedAgent = new HashMap<>();