Fix NPE when date_buckets aggregation is missing in the response (elastic#128974) (elastic#137337)

Author: przemekwitek

docs/changelog/128974.yaml

@@ -0,0 +1,5 @@
+pr: 128974
+summary: Fix NPE when `date_buckets` aggregation is missing in the response
+area: Machine Learning
+type: bug
+issues: []

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/delayeddatacheck/DatafeedDelayedDataDetector.java

@@ -6,6 +6,8 @@
  */
 package org.elasticsearch.xpack.ml.datafeed.delayeddatacheck;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.TransportSearchAction;
@@ -39,6 +41,8 @@
  */
 public class DatafeedDelayedDataDetector implements DelayedDataDetector {
 
+    private static final Logger logger = LogManager.getLogger(DatafeedDelayedDataDetector.class);
+
     private static final String DATE_BUCKETS = "date_buckets";
 
     private final long bucketSpan;
@@ -134,9 +138,16 @@ private Map<Long, Long> checkCurrentBucketEventCount(long start, long end) {
 
         SearchRequest searchRequest = new SearchRequest(datafeedIndices).source(searchSourceBuilder).indicesOptions(indicesOptions);
         try (ThreadContext.StoredContext ignore = client.threadPool().getThreadContext().stashWithOrigin(ML_ORIGIN)) {
-            SearchResponse response = client.execute(TransportSearchAction.TYPE, searchRequest).actionGet();
+            SearchResponse searchResponse = client.execute(TransportSearchAction.TYPE, searchRequest).actionGet();
             try {
-                List<? extends Histogram.Bucket> buckets = ((Histogram) response.getAggregations().get(DATE_BUCKETS)).getBuckets();
+                Histogram histogram = searchResponse.getAggregations().get(DATE_BUCKETS);
+                if (histogram == null) {
+                    // We log search response here to get information about shards and hits which may be helpful while debugging.
+                    // The size of the search response is small as we only log if the "date_buckets" aggregation is missing.
+                    logger.warn("[{}] Delayed data check failed with missing aggregation in search response [{}]", jobId, searchResponse);
+                    return Collections.emptyMap();
+                }
+                List<? extends Histogram.Bucket> buckets = histogram.getBuckets();
                 Map<Long, Long> hashMap = Maps.newMapWithExpectedSize(buckets.size());
                 for (Histogram.Bucket bucket : buckets) {
                     long bucketTime = toHistogramKeyToEpoch(bucket.getKey());
@@ -147,7 +158,7 @@ private Map<Long, Long> checkCurrentBucketEventCount(long start, long end) {
                 }
                 return hashMap;
             } finally {
-                response.decRef();
+                searchResponse.decRef();
             }
         }
     }