Resolve duplicate key exception in GetDatafeedRunningStateAction

Implement merge function for duplicate datafeed states when a datafeed is force-stopped and restarted before cancellation completes. Select the most appropriate state based on:
1. Prefer state with more recent searchInterval.startMs when both exist
2. Prefer states with searchInterval over those without
3. Default to second state when all criteria are equal

Author: hye-on

docs/changelog/125477.yaml

@@ -0,0 +1,6 @@
+pr: 125477
+summary: Prevent get datafeeds stats API returning an error when local tasks are slow to stop
+area: Machine Learning
+type: bug
+issues:
+  - 104160

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/GetDatafeedRunningStateAction.java

@@ -145,13 +145,28 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         }
 
         private final Map<String, RunningState> datafeedRunningState;
+        private static RunningState selectMostRecentState(RunningState state1, RunningState state2) {
 
+            if (state1.searchInterval != null && state2.searchInterval != null) {
+                return state1.searchInterval.startMs() > state2.searchInterval.startMs() ? state1 : state2;
+            }
+
+            if (state1.searchInterval != null) {
+                return state1;
+            }
+            if (state2.searchInterval != null) {
+                return state2;
+            }
+
+
+              return state2;
+        }
         public static Response fromResponses(List<Response> responses) {
             return new Response(
                 responses.stream()
                     .flatMap(r -> r.datafeedRunningState.entrySet().stream())
                     .filter(entry -> entry.getValue() != null)
-                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
+                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, Response::selectMostRecentState))
             );
         }