Dedup NodeOperationDetail across all traces in window

Move dedup map from per-trace to window level so the same
relationship from different traces produces only one document.

Signed-off-by: Vamsi Manohar <reddyvam@amazon.com>

Author: vamsimanohar

data-prepper-plugins/otel-apm-service-map-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/otel_apm_service_map/OTelApmServiceMapProcessor.java

@@ -416,6 +416,7 @@ private Collection<Record<Event>> processCurrentWindowSpans() {
 
         final Map<MetricKey, MetricAggregationState> sumStateByKey = new HashMap<>();
         final Map<MetricKey, MetricAggregationState> histogramStateByKey = new HashMap<>();
+        final Map<String, NodeOperationDetail> dedupedNodeDetails = new HashMap<>();
 
         final Map<String, Collection<SpanStateData>> previousSpansByTraceId = buildSpansByTraceIdMap(previousWindow);
         final Map<String, Collection<SpanStateData>> currentSpansByTraceId = buildSpansByTraceIdMap(currentWindow);
@@ -428,10 +429,23 @@ private Collection<Record<Event>> processCurrentWindowSpans() {
             if (!traceData.getProcessingSpans().isEmpty()) {
                 decorateSpansInTraceWithEphemeralStorage(traceData);
 
-                apmEvents.addAll(generateNodeOperationDetailEvents(traceData, currentTime, sumStateByKey, histogramStateByKey));
+                generateNodeOperationDetailEvents(traceData, currentTime, sumStateByKey, histogramStateByKey, dedupedNodeDetails);
             }
         }
 
+        // Convert deduped NodeOperationDetails to events
+        for (NodeOperationDetail detail : dedupedNodeDetails.values()) {
+            final EventMetadata eventMetadata = new DefaultEventMetadata.Builder()
+                    .withEventType(EVENT_TYPE_OTEL_APM_SERVICE_MAP).build();
+
+            final Event event = eventFactory.eventBuilder(EventBuilder.class)
+                    .withEventMetadata(eventMetadata)
+                    .withData(detail)
+                    .build();
+
+            apmEvents.add(new Record<>(event));
+        }
+
         final List<JacksonMetric> metrics = ApmServiceMapMetricsUtil.createMetricsFromAggregatedState(sumStateByKey, histogramStateByKey);
         metrics.sort(Comparator.comparing(JacksonMetric::getTime));
 
@@ -782,13 +796,11 @@ private void decorateServerSpansSecondPassWithEphemeralStorage(final ThreeWindow
      * @param metricsStateByKey Shared map for metric aggregation across all traces
      * @return Collection of NodeOperationDetail events
      */
-    private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeWindowTraceDataWithDecorations traceData,
-                                                                        final Instant currentTime,
-                                                                        final Map<MetricKey, MetricAggregationState> sumStateByKey,
-                                                                        final Map<MetricKey, MetricAggregationState> histogramStateByKey) {
-        // Dedup NodeOperationDetail by hash — first occurrence wins
-        final Map<String, NodeOperationDetail> dedupedNodeDetails = new HashMap<>();
-
+    private void generateNodeOperationDetailEvents(final ThreeWindowTraceDataWithDecorations traceData,
+                                                    final Instant currentTime,
+                                                    final Map<MetricKey, MetricAggregationState> sumStateByKey,
+                                                    final Map<MetricKey, MetricAggregationState> histogramStateByKey,
+                                                    final Map<String, NodeOperationDetail> dedupedNodeDetails) {
         // Step 1: CLIENT spans — primary emission path
         for (SpanStateData clientSpan : traceData.getProcessingSpans()) {
             if (SPAN_KIND_CLIENT.equals(clientSpan.getSpanKind())) {
@@ -859,22 +871,6 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
                 }
             }
         }
-
-        // Convert deduped NodeOperationDetails to events
-        final Collection<Record<Event>> events = new HashSet<>();
-        for (NodeOperationDetail detail : dedupedNodeDetails.values()) {
-            final EventMetadata eventMetadata = new DefaultEventMetadata.Builder()
-                    .withEventType(EVENT_TYPE_OTEL_APM_SERVICE_MAP).build();
-
-            final Event event = eventFactory.eventBuilder(EventBuilder.class)
-                    .withEventMetadata(eventMetadata)
-                    .withData(detail)
-                    .build();
-
-            events.add(new Record<>(event));
-        }
-
-        return events;
     }
 
     /**