Fixed timeout (de-)serialization for start and stop datafeeder and open job apis.

Original commit: elastic/x-pack-elasticsearch@be054db48c

elasticsearch/src/main/java/org/elasticsearch/xpack/ml/action/OpenJobAction.java

@@ -113,7 +113,7 @@ public class OpenJobAction extends Action<OpenJobAction.Request, OpenJobAction.R
 
         @Override
         public int hashCode() {
-            return Objects.hash(jobId, ignoreDowntime);
+            return Objects.hash(jobId, ignoreDowntime, openTimeout);
         }
 
         @Override
@@ -126,7 +126,8 @@ public class OpenJobAction extends Action<OpenJobAction.Request, OpenJobAction.R
             }
             OpenJobAction.Request other = (OpenJobAction.Request) obj;
             return Objects.equals(jobId, other.jobId) &&
-                    Objects.equals(ignoreDowntime, other.ignoreDowntime);
+                    Objects.equals(ignoreDowntime, other.ignoreDowntime) &&
+                    Objects.equals(openTimeout, other.openTimeout);
         }
     }
 

elasticsearch/src/main/java/org/elasticsearch/xpack/ml/action/StartDatafeedAction.java

@@ -30,12 +30,12 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.tasks.LoggingTaskListener;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
-import org.elasticsearch.xpack.ml.job.metadata.MlMetadata;
-import org.elasticsearch.xpack.ml.datafeed.DatafeedJobRunner;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedConfig;
+import org.elasticsearch.xpack.ml.datafeed.DatafeedJobRunner;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedStatus;
-import org.elasticsearch.xpack.ml.utils.ExceptionsHelper;
+import org.elasticsearch.xpack.ml.job.metadata.MlMetadata;
 import org.elasticsearch.xpack.ml.utils.DatafeedStatusObserver;
+import org.elasticsearch.xpack.ml.utils.ExceptionsHelper;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -152,13 +152,16 @@ public class StartDatafeedAction
             if (endTime != null) {
                 builder.field(END_TIME.getPreferredName(), endTime);
             }
+            if (startTimeout != null) {
+                builder.field(START_TIMEOUT.getPreferredName(), startTimeout.getStringRep());
+            }
             builder.endObject();
             return builder;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(datafeedId, startTime, endTime);
+            return Objects.hash(datafeedId, startTime, endTime, startTimeout);
         }
 
         @Override
@@ -172,7 +175,8 @@ public class StartDatafeedAction
             Request other = (Request) obj;
             return Objects.equals(datafeedId, other.datafeedId) &&
                     Objects.equals(startTime, other.startTime) &&
-                    Objects.equals(endTime, other.endTime);
+                    Objects.equals(endTime, other.endTime) &&
+                    Objects.equals(startTimeout, other.startTimeout);
         }
     }
 

elasticsearch/src/main/java/org/elasticsearch/xpack/ml/action/StopDatafeedAction.java

@@ -33,13 +33,13 @@ import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.tasks.TaskInfo;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
-import org.elasticsearch.xpack.ml.job.messages.Messages;
-import org.elasticsearch.xpack.ml.job.metadata.MlMetadata;
 import org.elasticsearch.xpack.ml.datafeed.Datafeed;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedConfig;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedStatus;
-import org.elasticsearch.xpack.ml.utils.ExceptionsHelper;
+import org.elasticsearch.xpack.ml.job.messages.Messages;
+import org.elasticsearch.xpack.ml.job.metadata.MlMetadata;
 import org.elasticsearch.xpack.ml.utils.DatafeedStatusObserver;
+import org.elasticsearch.xpack.ml.utils.ExceptionsHelper;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -93,17 +93,19 @@ public class StopDatafeedAction
         public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);
             datafeedId = in.readString();
+            stopTimeout = TimeValue.timeValueMillis(in.readVLong());
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeString(datafeedId);
+            out.writeVLong(stopTimeout.millis());
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(datafeedId);
+            return Objects.hash(datafeedId, stopTimeout);
         }
 
         @Override
@@ -115,7 +117,8 @@ public class StopDatafeedAction
                 return false;
             }
             Request other = (Request) obj;
-            return Objects.equals(datafeedId, other.datafeedId);
+            return Objects.equals(datafeedId, other.datafeedId) &&
+                    Objects.equals(stopTimeout, other.stopTimeout);
         }
     }
 

elasticsearch/src/test/java/org/elasticsearch/xpack/ml/action/CloseJobActionRequestTests.java

@@ -5,14 +5,19 @@
  */
 package org.elasticsearch.xpack.ml.action;
 
-import org.elasticsearch.xpack.ml.support.AbstractStreamableTestCase;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.xpack.ml.action.CloseJobAction.Request;
+import org.elasticsearch.xpack.ml.support.AbstractStreamableTestCase;
 
 public class CloseJobActionRequestTests extends AbstractStreamableTestCase<Request> {
 
     @Override
     protected Request createTestInstance() {
-        return new Request(randomAsciiOfLengthBetween(1, 20));
+        Request request = new Request(randomAsciiOfLengthBetween(1, 20));
+        if (randomBoolean()) {
+            request.setCloseTimeout(TimeValue.timeValueMillis(randomNonNegativeLong()));
+        }
+        return request;
     }
 
     @Override

elasticsearch/src/test/java/org/elasticsearch/xpack/ml/action/OpenJobActionRequestTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.ml.action;
 
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.xpack.ml.action.OpenJobAction.Request;
 import org.elasticsearch.xpack.ml.support.AbstractStreamableTestCase;
 
@@ -12,7 +13,11 @@ public class OpenJobActionRequestTests extends AbstractStreamableTestCase<Reques
 
     @Override
     protected Request createTestInstance() {
-        return new Request(randomAsciiOfLengthBetween(1, 20));
+        Request request = new Request(randomAsciiOfLengthBetween(1, 20));
+        if (randomBoolean()) {
+            request.setOpenTimeout(TimeValue.timeValueMillis(randomNonNegativeLong()));
+        }
+        return request;
     }
 
     @Override

elasticsearch/src/test/java/org/elasticsearch/xpack/ml/action/StartDatafeedActionRequestTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.ml.action;
 
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.xpack.ml.action.StartDatafeedAction.Request;
 import org.elasticsearch.xpack.ml.support.AbstractStreamableXContentTestCase;
@@ -17,6 +18,9 @@ public class StartDatafeedActionRequestTests extends AbstractStreamableXContentT
         if (randomBoolean()) {
             request.setEndTime(randomNonNegativeLong());
         }
+        if (randomBoolean()) {
+            request.setStartTimeout(TimeValue.timeValueMillis(randomNonNegativeLong()));
+        }
         return request;
     }
 

elasticsearch/src/test/java/org/elasticsearch/xpack/ml/action/StopDatafeedActionRequestTests.java

@@ -9,14 +9,14 @@ import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.xpack.ml.action.StopDatafeedAction.Request;
-import org.elasticsearch.xpack.ml.job.config.Job;
-import org.elasticsearch.xpack.ml.job.metadata.MlMetadata;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedConfig;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedStatus;
+import org.elasticsearch.xpack.ml.job.config.Job;
+import org.elasticsearch.xpack.ml.job.metadata.MlMetadata;
 import org.elasticsearch.xpack.ml.support.AbstractStreamableTestCase;
 
-import static org.elasticsearch.xpack.ml.datafeed.DatafeedJobRunnerTests.createDatafeedJob;
 import static org.elasticsearch.xpack.ml.datafeed.DatafeedJobRunnerTests.createDatafeedConfig;
+import static org.elasticsearch.xpack.ml.datafeed.DatafeedJobRunnerTests.createDatafeedJob;
 import static org.hamcrest.Matchers.equalTo;
 
 public class StopDatafeedActionRequestTests extends AbstractStreamableTestCase<StopDatafeedAction.Request> {
@@ -24,7 +24,9 @@ public class StopDatafeedActionRequestTests extends AbstractStreamableTestCase<S
     @Override
     protected Request createTestInstance() {
         Request r = new Request(randomAsciiOfLengthBetween(1, 20));
-        r.setStopTimeout(TimeValue.timeValueSeconds(randomIntBetween(0, 999)));
+        if (randomBoolean()) {
+            r.setStopTimeout(TimeValue.timeValueMillis(randomNonNegativeLong()));
+        }
         return r;
     }