fix(map_entries): Improve edge case handling for missing keys and empty filtered results

- Change missing source key from warn+tags_on_failure to silent debug
  log, consistent with other mutate-event processors
- Remove early return when all elements are filtered out by
  exclude_null_empty_values, ensuring consistent List<Map> output type
- Update test and README to reflect new behavior

Author: nishantKadivar

data-prepper-plugins/mutate-event-processors/README.md

@@ -749,10 +749,10 @@ After `add_entries`:
 * `tags_on_failure` - (optional) - A list of tags to add to the event metadata when the event fails to process
 
 ### Edge Case Behavior
-* If the `source` key does not exist in the event, the processor skips the event (no-op) and adds `tags_on_failure` if configured.
+* If the `source` key does not exist in the event, the processor silently skips the event (no-op). A debug-level log is emitted.
 * If the `source` value is not a list (e.g., a string or number), the processor skips the event (no-op) and adds `tags_on_failure` if configured.
 * If the `source` list is empty, the processor does nothing — the empty list remains as-is.
-* If no elements remain after filtering (when `exclude_null_empty_values` is `true` and all elements are null or empty), the original list is left unchanged.
+* If no elements remain after filtering (when `exclude_null_empty_values` is `true` and all elements are null or empty), the target is written as an empty list `[]`.
 * Null elements within the list are wrapped like any other value by default: `[null]` becomes `[{"key": null}]`. Use `exclude_null_empty_values: true` to filter them out.
 
 ___

data-prepper-plugins/mutate-event-processors/src/main/java/org/opensearch/dataprepper/plugins/processor/mutateevent/MapEntriesProcessor.java

@@ -20,8 +20,8 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -53,13 +53,6 @@ public MapEntriesProcessor(final PluginMetrics pluginMetrics,
                             "for valid expression syntax", config.getMapEntriesWhen()));
         }
 
-        if (config.getTagsOnFailure() != null) {
-            for (final String tag : config.getTagsOnFailure()) {
-                if (tag == null || tag.isEmpty()) {
-                    throw new InvalidPluginConfigurationException("tags_on_failure must not contain null or empty strings.");
-                }
-            }
-        }
     }
 
     @Override
@@ -92,8 +85,7 @@ private void processEvent(final Event event) {
         final String source = config.getSource();
 
         if (!event.containsKey(source)) {
-            LOG.warn(EVENT, "Source key [{}] does not exist in event [{}], skipping.", source, event);
-            addTagsOnFailure(event);
+            LOG.debug(EVENT, "Source key [{}] does not exist in event [{}], skipping.", source, event);
             return;
         }
 
@@ -123,14 +115,9 @@ private void processEvent(final Event event) {
                     continue;
                 }
             }
-            final Map<String, Object> wrapped = new HashMap<>(1);
-            wrapped.put(key, element);
-            result.add(wrapped);
+            result.add(Collections.singletonMap(key, element));
         }
 
-        if (result.isEmpty()) {
-            return;
-        }
 
         final String effectiveTarget = config.getEffectiveTarget();
 

data-prepper-plugins/mutate-event-processors/src/test/java/org/opensearch/dataprepper/plugins/processor/mutateevent/MapEntriesProcessorTest.java

@@ -96,21 +96,6 @@ void constructor_with_empty_target_throws_InvalidPluginConfigurationException()
         assertThrows(InvalidPluginConfigurationException.class, this::createObjectUnderTest);
     }
 
-    @Test
-    void constructor_with_empty_string_in_tags_on_failure_throws_InvalidPluginConfigurationException() {
-        when(config.getTagsOnFailure()).thenReturn(List.of(""));
-
-        assertThrows(InvalidPluginConfigurationException.class, this::createObjectUnderTest);
-    }
-
-    @Test
-    void constructor_with_null_element_in_tags_on_failure_throws_InvalidPluginConfigurationException() {
-        final List<String> tags = new ArrayList<>();
-        tags.add(null);
-        when(config.getTagsOnFailure()).thenReturn(tags);
-
-        assertThrows(InvalidPluginConfigurationException.class, this::createObjectUnderTest);
-    }
 
     // --- Basic wrapping ---
 
@@ -192,7 +177,7 @@ void doExecute_with_exclude_enabled_filters_out_null_and_empty_elements() {
     }
 
     @Test
-    void doExecute_with_all_null_empty_and_exclude_enabled_leaves_original_list_unchanged() {
+    void doExecute_with_all_null_empty_and_exclude_enabled_writes_empty_list() {
         when(config.getExcludeNullEmptyValues()).thenReturn(true);
 
         final List<String> input = new ArrayList<>();
@@ -204,7 +189,7 @@ void doExecute_with_all_null_empty_and_exclude_enabled_leaves_original_list_unch
 
         final Object output = result.get(0).getData().get("/names", Object.class);
         assertThat(output instanceof List, is(true));
-        assertThat(((List<?>) output).size(), is(2));
+        assertThat(((List<?>) output).size(), is(0));
     }
 
     // --- Skip conditions ---