Adapt for origin setting client

Author: davidkyle

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/retention/ExpiredModelSnapshotsRemover.java

@@ -42,9 +42,6 @@
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
-import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
-import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
-
 /**
  * Deletes all model snapshots that have expired the configured retention time
  * of their respective job with the exception of the currently used snapshot.
@@ -109,19 +106,19 @@ private void latestSnapshotTimeStamp(String jobId, ActionListener<Long> listener
         searchRequest.source(searchSourceBuilder);
         searchRequest.indicesOptions(MlIndicesUtils.addIgnoreUnavailable(SearchRequest.DEFAULT_INDICES_OPTIONS));
 
-        executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, searchRequest,
-                ActionListener.<SearchResponse>wrap(
-                        response -> {
-                            SearchHit[] hits = response.getHits().getHits();
-                            if (hits.length == 0) {
-                                // no snapshots found
-                                listener.onResponse(null);
-                            } else {
-                                ModelSnapshot snapshot = ModelSnapshot.fromJson(hits[0].getSourceRef());
-                                listener.onResponse(snapshot.getTimestamp().getTime());
-                            }
-                        }, listener::onFailure
-                ), client::search);
+        client.search(searchRequest, ActionListener.wrap(
+                response -> {
+                    SearchHit[] hits = response.getHits().getHits();
+                    if (hits.length == 0) {
+                        // no snapshots found
+                        listener.onResponse(null);
+                    } else {
+                        ModelSnapshot snapshot = ModelSnapshot.fromJson(hits[0].getSourceRef());
+                        listener.onResponse(snapshot.getTimestamp().getTime());
+                    }
+                },
+                listener::onFailure)
+        );
     }
 
     @Override

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/retention/ExpiredResultsRemover.java

@@ -11,7 +11,6 @@
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchRequest;
-import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.support.ThreadedActionListener;
 import org.elasticsearch.client.OriginSettingClient;
 import org.elasticsearch.common.unit.TimeValue;
@@ -54,9 +53,6 @@
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
-import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
-import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
-
 /**
  * Removes all results that have expired the configured retention time
  * of their respective job. A result is deleted if its timestamp is earlier
@@ -162,26 +158,25 @@ private void latestBucketTime(String jobId, ActionListener<Long> listener) {
         searchRequest.source(searchSourceBuilder);
         searchRequest.indicesOptions(MlIndicesUtils.addIgnoreUnavailable(SearchRequest.DEFAULT_INDICES_OPTIONS));
 
-        executeAsyncWithOrigin(client.threadPool().getThreadContext(), ML_ORIGIN, searchRequest,
-                ActionListener.<SearchResponse>wrap(
-                        response -> {
-                            SearchHit[] hits = response.getHits().getHits();
-                            if (hits.length == 0) {
-                                // no buckets found
-                                listener.onResponse(null);
-                            } else {
-
-                                try (InputStream stream = hits[0].getSourceRef().streamInput();
-                                     XContentParser parser = XContentFactory.xContent(XContentType.JSON)
-                                             .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, stream)) {
-                                    Bucket bucket = Bucket.LENIENT_PARSER.apply(parser, null);
-                                    listener.onResponse(bucket.getTimestamp().getTime());
-                                } catch (IOException e) {
-                                    listener.onFailure(new ElasticsearchParseException("failed to parse bucket", e));
-                                }
-                            }
-                        }, listener::onFailure
-                ), client::search);
+        client.search(searchRequest, ActionListener.wrap(
+                response -> {
+                    SearchHit[] hits = response.getHits().getHits();
+                    if (hits.length == 0) {
+                        // no buckets found
+                        listener.onResponse(null);
+                    } else {
+
+                        try (InputStream stream = hits[0].getSourceRef().streamInput();
+                             XContentParser parser = XContentFactory.xContent(XContentType.JSON)
+                                     .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, stream)) {
+                            Bucket bucket = Bucket.LENIENT_PARSER.apply(parser, null);
+                            listener.onResponse(bucket.getTimestamp().getTime());
+                        } catch (IOException e) {
+                            listener.onFailure(new ElasticsearchParseException("failed to parse bucket", e));
+                        }
+                    }
+                }, listener::onFailure
+        ));
     }
 
     private void auditResultsWereDeleted(String jobId, long cutoffEpochMs) {