[ML] Return assigned node in start/open job/datafeed response

client/rest-high-level/src/main/java/org/elasticsearch/client/MachineLearningClient.java

@@ -108,6 +108,7 @@
 import org.elasticsearch.client.ml.RevertModelSnapshotResponse;
 import org.elasticsearch.client.ml.SetUpgradeModeRequest;
 import org.elasticsearch.client.ml.StartDataFrameAnalyticsRequest;
+import org.elasticsearch.client.ml.StartDataFrameAnalyticsResponse;
 import org.elasticsearch.client.ml.StartDatafeedRequest;
 import org.elasticsearch.client.ml.StartDatafeedResponse;
 import org.elasticsearch.client.ml.StopDataFrameAnalyticsRequest;
@@ -2138,12 +2139,12 @@ public Cancellable getDataFrameAnalyticsStatsAsync(GetDataFrameAnalyticsStatsReq
      * @return action acknowledgement
      * @throws IOException when there is a serialization issue sending the request or receiving the response
      */
-    public AcknowledgedResponse startDataFrameAnalytics(StartDataFrameAnalyticsRequest request,
-                                                        RequestOptions options) throws IOException {
+    public StartDataFrameAnalyticsResponse startDataFrameAnalytics(StartDataFrameAnalyticsRequest request,
+                                                                   RequestOptions options) throws IOException {
         return restHighLevelClient.performRequestAndParseEntity(request,
             MLRequestConverters::startDataFrameAnalytics,
             options,
-            AcknowledgedResponse::fromXContent,
+            StartDataFrameAnalyticsResponse::fromXContent,
             Collections.emptySet());
     }
 
@@ -2160,11 +2161,11 @@ public AcknowledgedResponse startDataFrameAnalytics(StartDataFrameAnalyticsReque
      * @return cancellable that may be used to cancel the request
      */
     public Cancellable startDataFrameAnalyticsAsync(StartDataFrameAnalyticsRequest request, RequestOptions options,
-                                                    ActionListener<AcknowledgedResponse> listener) {
+                                                    ActionListener<StartDataFrameAnalyticsResponse> listener) {
         return restHighLevelClient.performRequestAsyncAndParseEntity(request,
             MLRequestConverters::startDataFrameAnalytics,
             options,
-            AcknowledgedResponse::fromXContent,
+            StartDataFrameAnalyticsResponse::fromXContent,
             listener,
             Collections.emptySet());
     }

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/OpenJobResponse.java

@@ -33,18 +33,23 @@
 public class OpenJobResponse implements ToXContentObject {
 
     private static final ParseField OPENED = new ParseField("opened");
+    private static final ParseField NODE = new ParseField("node");
 
     public static final ConstructingObjectParser<OpenJobResponse, Void> PARSER =
-        new ConstructingObjectParser<>("open_job_response", true, (a) -> new OpenJobResponse((Boolean)a[0]));
+        new ConstructingObjectParser<>("open_job_response", true,
+            (a) -> new OpenJobResponse((Boolean) a[0], (String) a[1]));
 
     static {
         PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), OPENED);
+        PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), NODE);
     }
 
-    private boolean opened;
+    private final boolean opened;
+    private final String node;
 
-    OpenJobResponse(boolean opened) {
+    OpenJobResponse(boolean opened, String node) {
         this.opened = opened;
+        this.node = node;
     }
 
     public static OpenJobResponse fromXContent(XContentParser parser) throws IOException {
@@ -60,6 +65,18 @@ public boolean isOpened() {
         return opened;
     }
 
+    /**
+     * The node that the job was assigned to
+     *
+     * @return The ID of a node if the job was assigned to a node.  If an empty string is returned
+     *         it means the job was allowed to open lazily and has not yet been assigned to a node.
+     *         If <code>null</code> is returned it means the server version is too old to return node
+     *         information.
+     */
+    public String getNode() {
+        return node;
+    }
+
     @Override
     public boolean equals(Object other) {
         if (this == other) {
@@ -71,18 +88,22 @@ public boolean equals(Object other) {
         }
 
         OpenJobResponse that = (OpenJobResponse) other;
-        return isOpened() == that.isOpened();
+        return opened == that.opened
+            && Objects.equals(node, that.node);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(isOpened());
+        return Objects.hash(opened, node);
     }
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         builder.field(OPENED.getPreferredName(), opened);
+        if (node != null) {
+            builder.field(NODE.getPreferredName(), node);
+        }
         builder.endObject();
         return builder;
     }

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/StartDataFrameAnalyticsResponse.java

@@ -0,0 +1,97 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Response indicating if the Machine Learning Datafeed is now started or not
+ */
+public class StartDataFrameAnalyticsResponse extends AcknowledgedResponse {
+
+    private static final ParseField NODE = new ParseField("node");
+
+    public static final ConstructingObjectParser<StartDataFrameAnalyticsResponse, Void> PARSER =
+        new ConstructingObjectParser<>(
+            "start_data_frame_analytics_response",
+            true,
+            (a) -> new StartDataFrameAnalyticsResponse((Boolean) a[0], (String) a[1]));
+
+    static {
+        declareAcknowledgedField(PARSER);
+        PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), NODE);
+    }
+
+    private final String node;
+
+    public StartDataFrameAnalyticsResponse(boolean acknowledged, String node) {
+        super(acknowledged);
+        this.node = node;
+    }
+
+    public static StartDataFrameAnalyticsResponse fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+
+    /**
+     * The node that the job was assigned to
+     *
+     * @return The ID of a node if the job was assigned to a node.  If an empty string is returned
+     *         it means the job was allowed to open lazily and has not yet been assigned to a node.
+     *         If <code>null</code> is returned it means the server version is too old to return node
+     *         information.
+     */
+    public String getNode() {
+        return node;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (this == other) {
+            return true;
+        }
+
+        if (other == null || getClass() != other.getClass()) {
+            return false;
+        }
+
+        StartDataFrameAnalyticsResponse that = (StartDataFrameAnalyticsResponse) other;
+        return isAcknowledged() == that.isAcknowledged()
+            && Objects.equals(node, that.node);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(isAcknowledged(), node);
+    }
+
+    @Override
+    public void addCustomFields(XContentBuilder builder, Params params) throws IOException {
+        if (node != null) {
+            builder.field(NODE.getPreferredName(), node);
+        }
+    }
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/StartDatafeedResponse.java

@@ -33,21 +33,25 @@
 public class StartDatafeedResponse implements ToXContentObject {
 
     private static final ParseField STARTED = new ParseField("started");
+    private static final ParseField NODE = new ParseField("node");
 
     public static final ConstructingObjectParser<StartDatafeedResponse, Void> PARSER =
         new ConstructingObjectParser<>(
             "start_datafeed_response",
             true,
-            (a) -> new StartDatafeedResponse((Boolean)a[0]));
+            (a) -> new StartDatafeedResponse((Boolean) a[0], (String) a[1]));
 
     static {
         PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), STARTED);
+        PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), NODE);
     }
 
     private final boolean started;
+    private final String node;
 
-    public StartDatafeedResponse(boolean started) {
+    public StartDatafeedResponse(boolean started, String node) {
         this.started = started;
+        this.node = node;
     }
 
     public static StartDatafeedResponse fromXContent(XContentParser parser) throws IOException {
@@ -63,6 +67,18 @@ public boolean isStarted() {
         return started;
     }
 
+    /**
+     * The node that the datafeed was assigned to
+     *
+     * @return The ID of a node if the datafeed was assigned to a node.  If an empty string is returned
+     *         it means the datafeed was allowed to open lazily and has not yet been assigned to a node.
+     *         If <code>null</code> is returned it means the server version is too old to return node
+     *         information.
+     */
+    public String getNode() {
+        return node;
+    }
+
     @Override
     public boolean equals(Object other) {
         if (this == other) {
@@ -74,18 +90,22 @@ public boolean equals(Object other) {
         }
 
         StartDatafeedResponse that = (StartDatafeedResponse) other;
-        return isStarted() == that.isStarted();
+        return started == started
+            && Objects.equals(node, that.node);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(isStarted());
+        return Objects.hash(started, node);
     }
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         builder.field(STARTED.getPreferredName(), started);
+        if (node != null) {
+            builder.field(NODE.getPreferredName(), node);
+        }
         builder.endObject();
         return builder;
     }

client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/MlClientDocumentationIT.java

@@ -122,6 +122,7 @@
 import org.elasticsearch.client.ml.RevertModelSnapshotResponse;
 import org.elasticsearch.client.ml.SetUpgradeModeRequest;
 import org.elasticsearch.client.ml.StartDataFrameAnalyticsRequest;
+import org.elasticsearch.client.ml.StartDataFrameAnalyticsResponse;
 import org.elasticsearch.client.ml.StartDatafeedRequest;
 import org.elasticsearch.client.ml.StartDatafeedResponse;
 import org.elasticsearch.client.ml.StopDataFrameAnalyticsRequest;
@@ -232,6 +233,7 @@
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.core.Is.is;
 
 public class MlClientDocumentationIT extends ESRestHighLevelClientTestCase {
@@ -461,7 +463,10 @@ public void testOpenJob() throws Exception {
 
             // tag::open-job-response
             boolean isOpened = openJobResponse.isOpened(); // <1>
+            String node = openJobResponse.getNode(); // <2>
             // end::open-job-response
+
+            assertThat(node, notNullValue());
         }
         {
             // tag::open-job-execute-listener
@@ -1011,11 +1016,14 @@ public void testStartDatafeed() throws Exception {
             // tag::start-datafeed-execute
             StartDatafeedResponse response = client.machineLearning().startDatafeed(request, RequestOptions.DEFAULT);
             // end::start-datafeed-execute
+
             // tag::start-datafeed-response
             boolean started = response.isStarted(); // <1>
+            String node = response.getNode(); // <2>
             // end::start-datafeed-response
 
             assertTrue(started);
+            assertThat(node, notNullValue());
         }
         {
             StartDatafeedRequest request = new StartDatafeedRequest(datafeedId);
@@ -3131,14 +3139,16 @@ public void testStartDataFrameAnalytics() throws Exception {
             // end::start-data-frame-analytics-request
 
             // tag::start-data-frame-analytics-execute
-            AcknowledgedResponse response = client.machineLearning().startDataFrameAnalytics(request, RequestOptions.DEFAULT);
+            StartDataFrameAnalyticsResponse response = client.machineLearning().startDataFrameAnalytics(request, RequestOptions.DEFAULT);
             // end::start-data-frame-analytics-execute
 
             // tag::start-data-frame-analytics-response
             boolean acknowledged = response.isAcknowledged();
+            String node = response.getNode(); // <1>
             // end::start-data-frame-analytics-response
 
             assertThat(acknowledged, is(true));
+            assertThat(node, notNullValue());
         }
         assertBusy(
             () -> assertThat(getAnalyticsState(DF_ANALYTICS_CONFIG.getId()), equalTo(DataFrameAnalyticsState.STOPPED)),
@@ -3147,9 +3157,9 @@ public void testStartDataFrameAnalytics() throws Exception {
             StartDataFrameAnalyticsRequest request = new StartDataFrameAnalyticsRequest("my-analytics-config");
 
             // tag::start-data-frame-analytics-execute-listener
-            ActionListener<AcknowledgedResponse> listener = new ActionListener<>() {
+            ActionListener<StartDataFrameAnalyticsResponse> listener = new ActionListener<>() {
                 @Override
-                public void onResponse(AcknowledgedResponse response) {
+                public void onResponse(StartDataFrameAnalyticsResponse response) {
                     // <1>
                 }
 

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/OpenJobResponseTests.java

@@ -27,7 +27,8 @@ public class OpenJobResponseTests extends AbstractXContentTestCase<OpenJobRespon
 
     @Override
     protected OpenJobResponse createTestInstance() {
-        return new OpenJobResponse(randomBoolean());
+        String node = randomFrom("", randomAlphaOfLength(10), null);
+        return new OpenJobResponse(randomBoolean(), node);
     }
 
     @Override
@@ -37,6 +38,6 @@ protected OpenJobResponse doParseInstance(XContentParser parser) throws IOExcept
 
     @Override
     protected boolean supportsUnknownFields() {
-        return false;
+        return true;
     }
 }