Expose timeout of acknowledged requests in REST layer (elastic/x-pack-elasticsearch#2259)

Companion PR to elastic/elasticsearch#26189

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

plugin/src/main/java/org/elasticsearch/license/PutLicenseRequest.java

@@ -66,7 +66,6 @@ public class PutLicenseRequest extends AcknowledgedRequest<PutLicenseRequest> {
         super.readFrom(in);
         license = License.readLicense(in);
         acknowledge = in.readBoolean();
-        readTimeout(in);
     }
 
     @Override
@@ -74,6 +73,5 @@ public class PutLicenseRequest extends AcknowledgedRequest<PutLicenseRequest> {
         super.writeTo(out);
         license.writeTo(out);
         out.writeBoolean(acknowledge);
-        writeTimeout(out);
     }
 }

plugin/src/main/java/org/elasticsearch/license/RestDeleteLicenseAction.java

@@ -29,9 +29,12 @@ public class RestDeleteLicenseAction extends XPackRestHandler {
 
     @Override
     public RestChannelConsumer doPrepareRequest(final RestRequest request, final XPackClient client) throws IOException {
-        return channel -> client.es().admin().cluster().execute(DeleteLicenseAction.INSTANCE,
+        DeleteLicenseRequest deleteLicenseRequest = new DeleteLicenseRequest();
-                                              new DeleteLicenseRequest(),
+        deleteLicenseRequest.timeout(request.paramAsTime("timeout", deleteLicenseRequest.timeout()));
-                                              new AcknowledgedRestListener<>(channel));
+        deleteLicenseRequest.masterNodeTimeout(request.paramAsTime("master_timeout", deleteLicenseRequest.masterNodeTimeout()));
+
+        return channel -> client.es().admin().cluster().execute(DeleteLicenseAction.INSTANCE, deleteLicenseRequest,
+                new AcknowledgedRestListener<>(channel));
     }
 
 }

plugin/src/main/java/org/elasticsearch/license/RestPutLicenseAction.java

@@ -44,6 +44,9 @@ public class RestPutLicenseAction extends XPackRestHandler {
         PutLicenseRequest putLicenseRequest = new PutLicenseRequest();
         putLicenseRequest.license(request.content(), request.getXContentType());
         putLicenseRequest.acknowledge(request.paramAsBoolean("acknowledge", false));
+        putLicenseRequest.timeout(request.paramAsTime("timeout", putLicenseRequest.timeout()));
+        putLicenseRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putLicenseRequest.masterNodeTimeout()));
+
         return channel -> client.es().admin().cluster().execute(PutLicenseAction.INSTANCE, putLicenseRequest,
                 new RestBuilderListener<PutLicenseResponse>(channel) {
                     @Override

plugin/src/main/java/org/elasticsearch/xpack/ml/rest/datafeeds/RestDeleteDatafeedAction.java

@@ -38,6 +38,8 @@ public class RestDeleteDatafeedAction extends BaseRestHandler {
         if (restRequest.hasParam(DeleteDatafeedAction.Request.FORCE.getPreferredName())) {
             request.setForce(restRequest.paramAsBoolean(CloseJobAction.Request.FORCE.getPreferredName(), request.isForce()));
         }
+        request.timeout(restRequest.paramAsTime("timeout", request.timeout()));
+        request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
         return channel -> client.execute(DeleteDatafeedAction.INSTANCE, request, new AcknowledgedRestListener<>(channel));
     }
 }

plugin/src/main/java/org/elasticsearch/xpack/ml/rest/datafeeds/RestPutDatafeedAction.java

@@ -36,6 +36,8 @@ public class RestPutDatafeedAction extends BaseRestHandler {
         String datafeedId = restRequest.param(DatafeedConfig.ID.getPreferredName());
         XContentParser parser = restRequest.contentParser();
         PutDatafeedAction.Request putDatafeedRequest = PutDatafeedAction.Request.parseRequest(datafeedId, parser);
+        putDatafeedRequest.timeout(restRequest.paramAsTime("timeout", putDatafeedRequest.timeout()));
+        putDatafeedRequest.masterNodeTimeout(restRequest.paramAsTime("master_timeout", putDatafeedRequest.masterNodeTimeout()));
         return channel -> client.execute(PutDatafeedAction.INSTANCE, putDatafeedRequest, new RestToXContentListener<>(channel));
     }
 

plugin/src/main/java/org/elasticsearch/xpack/ml/rest/datafeeds/RestUpdateDatafeedAction.java

@@ -36,6 +36,9 @@ public class RestUpdateDatafeedAction extends BaseRestHandler {
         String datafeedId = restRequest.param(DatafeedConfig.ID.getPreferredName());
         XContentParser parser = restRequest.contentParser();
         UpdateDatafeedAction.Request updateDatafeedRequest = UpdateDatafeedAction.Request.parseRequest(datafeedId, parser);
+        updateDatafeedRequest.timeout(restRequest.paramAsTime("timeout", updateDatafeedRequest.timeout()));
+        updateDatafeedRequest.masterNodeTimeout(restRequest.paramAsTime("master_timeout", updateDatafeedRequest.masterNodeTimeout()));
+
         return channel -> client.execute(UpdateDatafeedAction.INSTANCE, updateDatafeedRequest, new RestToXContentListener<>(channel));
     }
 

plugin/src/main/java/org/elasticsearch/xpack/ml/rest/filter/RestDeleteFilterAction.java

@@ -33,6 +33,8 @@ public class RestDeleteFilterAction extends BaseRestHandler {
     @Override
     protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
         Request request = new Request(restRequest.param(Request.FILTER_ID.getPreferredName()));
+        request.timeout(restRequest.paramAsTime("timeout", request.timeout()));
+        request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
         return channel -> client.execute(DeleteFilterAction.INSTANCE, request, new AcknowledgedRestListener<>(channel));
     }
 

plugin/src/main/java/org/elasticsearch/xpack/ml/rest/job/RestDeleteJobAction.java

@@ -35,6 +35,8 @@ public class RestDeleteJobAction extends BaseRestHandler {
     protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
         DeleteJobAction.Request deleteJobRequest = new DeleteJobAction.Request(restRequest.param(Job.ID.getPreferredName()));
         deleteJobRequest.setForce(restRequest.paramAsBoolean(CloseJobAction.Request.FORCE.getPreferredName(), deleteJobRequest.isForce()));
+        deleteJobRequest.timeout(restRequest.paramAsTime("timeout", deleteJobRequest.timeout()));
+        deleteJobRequest.masterNodeTimeout(restRequest.paramAsTime("master_timeout", deleteJobRequest.masterNodeTimeout()));
         return channel -> client.execute(DeleteJobAction.INSTANCE, deleteJobRequest, new AcknowledgedRestListener<>(channel));
     }
 }

plugin/src/main/java/org/elasticsearch/xpack/ml/rest/job/RestPostJobUpdateAction.java

@@ -35,6 +35,9 @@ public class RestPostJobUpdateAction extends BaseRestHandler {
         String jobId = restRequest.param(Job.ID.getPreferredName());
         XContentParser parser = restRequest.contentParser();
         UpdateJobAction.Request updateJobRequest = UpdateJobAction.Request.parseRequest(jobId, parser);
+        updateJobRequest.timeout(restRequest.paramAsTime("timeout", updateJobRequest.timeout()));
+        updateJobRequest.masterNodeTimeout(restRequest.paramAsTime("master_timeout", updateJobRequest.masterNodeTimeout()));
+
         return channel -> client.execute(UpdateJobAction.INSTANCE, updateJobRequest, new RestToXContentListener<>(channel));
     }
 }

plugin/src/main/java/org/elasticsearch/xpack/ml/rest/job/RestPutJobAction.java

@@ -36,6 +36,9 @@ public class RestPutJobAction extends BaseRestHandler {
         String jobId = restRequest.param(Job.ID.getPreferredName());
         XContentParser parser = restRequest.contentParser();
         PutJobAction.Request putJobRequest = PutJobAction.Request.parseRequest(jobId, parser);
+        putJobRequest.timeout(restRequest.paramAsTime("timeout", putJobRequest.timeout()));
+        putJobRequest.masterNodeTimeout(restRequest.paramAsTime("master_timeout", putJobRequest.masterNodeTimeout()));
+
         return channel -> client.execute(PutJobAction.INSTANCE, putJobRequest, new RestToXContentListener<>(channel));
     }
 

plugin/src/main/java/org/elasticsearch/xpack/ml/rest/modelsnapshots/RestRevertModelSnapshotAction.java

@@ -47,6 +47,9 @@ public class RestRevertModelSnapshotAction extends BaseRestHandler {
             request.setDeleteInterveningResults(restRequest
                     .paramAsBoolean(RevertModelSnapshotAction.Request.DELETE_INTERVENING.getPreferredName(), DELETE_INTERVENING_DEFAULT));
         }
+        request.timeout(restRequest.paramAsTime("timeout", request.timeout()));
+        request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
+
         return channel -> client.execute(RevertModelSnapshotAction.INSTANCE, request, new RestStatusToXContentListener<>(channel));
     }
 }

plugin/src/test/java/org/elasticsearch/xpack/monitoring/action/MonitoringBulkDocTests.java

@@ -16,7 +16,6 @@ import org.elasticsearch.xpack.monitoring.MonitoredSystem;
 import org.elasticsearch.xpack.monitoring.exporter.MonitoringDoc;
 
 import java.io.IOException;
-import java.util.Base64;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -56,29 +55,6 @@ public class MonitoringBulkDocTests extends ESTestCase {
         }
     }
 
-    public void testSerializationBwc() throws IOException {
-        final byte[] data = Base64.getDecoder().decode("AQNtSWQBBTUuMS4yAAAAAQEEdHlwZQECaWQNeyJmb28iOiJiYXIifQAAAAAAAAAA");
-        final Version version = randomFrom(Version.V_5_0_0, Version.V_5_0_1, Version.V_5_0_2,
-                Version.V_5_1_1, Version.V_5_1_2, Version.V_5_2_0);
-        try (StreamInput in = StreamInput.wrap(data)) {
-            in.setVersion(version);
-            MonitoringBulkDoc bulkDoc = MonitoringBulkDoc.readFrom(in);
-            assertEquals(XContentType.JSON, bulkDoc.getXContentType());
-            assertEquals("mId", bulkDoc.getMonitoringId());
-            assertEquals("5.1.2", bulkDoc.getMonitoringVersion());
-            assertEquals(MonitoringIndex.TIMESTAMPED, bulkDoc.getIndex());
-            assertEquals("{\"foo\":\"bar\"}", bulkDoc.getSource().utf8ToString());
-            assertEquals("type", bulkDoc.getType());
-            assertEquals("id", bulkDoc.getId());
-
-            try (BytesStreamOutput out = new BytesStreamOutput()) {
-                out.setVersion(version);
-                bulkDoc.writeTo(out);
-                assertArrayEquals(data, out.bytes().toBytesRef().bytes);
-            }
-        }
-    }
-
     public static MonitoringBulkDoc newRandomMonitoringBulkDoc() {
         String monitoringId = randomFrom(MonitoredSystem.values()).getSystem();
         String monitoringVersion = randomVersion(random()).toString();

plugin/src/test/java/org/elasticsearch/xpack/watcher/transport/action/execute/ExecuteWatchRequestTests.java

@@ -5,7 +5,6 @@
  */
 package org.elasticsearch.xpack.watcher.transport.action.execute;
 
-import org.elasticsearch.Version;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -15,7 +14,6 @@ import org.elasticsearch.xpack.watcher.transport.actions.execute.ExecuteWatchReq
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.Base64;
 
 public class ExecuteWatchRequestTests extends ESTestCase {
 
@@ -32,23 +30,4 @@ public class ExecuteWatchRequestTests extends ESTestCase {
         assertEquals(XContentType.JSON, serialized.getXContentType());
         assertEquals("{}", serialized.getWatchSource().utf8ToString());
     }
-
-    public void testSerializationBwc() throws IOException {
-        final byte[] data = Base64.getDecoder().decode("ADwDAAAAAAAAAAAAAAAAAAABDnsid2F0Y2giOiJtZSJ9AAAAAAAAAA==");
-        final Version version = randomFrom(Version.V_5_0_0, Version.V_5_0_1, Version.V_5_0_2,
-                Version.V_5_1_1, Version.V_5_1_2, Version.V_5_2_0);
-        try (StreamInput in = StreamInput.wrap(data)) {
-            in.setVersion(version);
-            ExecuteWatchRequest request = new ExecuteWatchRequest();
-            request.readFrom(in);
-            assertEquals(XContentType.JSON, request.getXContentType());
-            assertEquals("{\"watch\":\"me\"}", request.getWatchSource().utf8ToString());
-
-            try (BytesStreamOutput out = new BytesStreamOutput()) {
-                out.setVersion(version);
-                request.writeTo(out);
-                assertArrayEquals(data, out.bytes().toBytesRef().bytes);
-            }
-        }
-    }
 }