Merge branch 'master' into feature/sql

Original commit: elastic/x-pack-elasticsearch@81ed649a9b

docs/en/commands/certgen.asciidoc

@@ -101,125 +101,6 @@ which can be blank. This parameter cannot be used with the `-csr` parameter.
 [float]
 === Examples
 
-////
-The tool can be used interactively:
-
-[source,shell]
---------------------------------------------------
-bin/x-pack/certgen
---------------------------------------------------
-This tool assists you in the generation of X.509 certificates and certificate
-signing requests for use with SSL in the Elastic stack. Depending on the command
-line option specified, you may be prompted for the following:
-
-* The path to the output file
-    * The output file is a zip file containing the signed certificates and
-      private keys for each instance. If a Certificate Authority was generated,
-      the certificate and private key will also be included in the output file.
-* Information about each instance
-    * An instance is any piece of the Elastic Stack that requires a SSL certificate.
-      Depending on your configuration, Elasticsearch, Logstash, Kibana, and Beats
-      may all require a certificate and private key.
-    * The minimum required value for each instance is a name. This can simply be the
-      hostname, which will be used as the Common Name of the certificate. A full
-      distinguished name may also be used.
-    * IP addresses and DNS names are optional. Multiple values can be specified as a
-      comma separated string. If no IP addresses or DNS names are provided, you may
-      disable hostname verification in your SSL configuration.
-* Certificate Authority private key password
-    * The password may be left empty if desired.
-
-Let's get started...
-
-Please enter the desired output file [/home/es/config/x-pack/certificate-bundle.zip]:
-Enter instance name: node01
-Enter name for directories and files [node01]:
-Enter IP Addresses for instance (comma-separated if more than one) []: 10.10.0.1
-Enter DNS names for instance (comma-separated if more than one) []: node01.mydomain.com,node01
-Would you like to specify another instance? Press 'y' to continue entering instance information: y
-Enter instance name: node02
-Enter name for directories and files [node02]:
-Enter IP Addresses for instance (comma-separated if more than one) []: 10.10.0.2
-Enter DNS names for instance (comma-separated if more than one) []: node02.mydomain.com
-Would you like to specify another instance? Press 'y' to continue entering instance information:
-Certificates written to /home/es/config/x-pack/certificate-bundle.zip
-
-This file should be properly secured as it contains the private keys for all
-instances and the certificate authority.
-
-After unzipping the file, there will be a directory for each instance containing
-the certificate and private key. Copy the certificate, key, and CA certificate
-to the configuration directory of the Elastic product that they will be used for
-and follow the SSL configuration instructions in the product guide.
-
-For client applications, you may only need to copy the CA certificate and
-configure the client to trust this certificate.
-....
---------------------------------------------------
-
-In this example, the command generates a zip file with the CA certificate,
-private key, two signed certificates and keys in PEM format for `node01` and
-`node02`.
-////
-////
-When using a commercial or organization specific CA, the `certgen` tool can be
-used to generate certificate signing requests (CSR) for the nodes in your
-cluster:
-
-[source,shell]
---------------------------------------------------
-....
-bin/x-pack/certgen -csr
-This tool assists you in the generation of X.509 certificates and certificate
-signing requests for use with SSL in the Elastic stack. Depending on the command
-line option specified, you may be prompted for the following:
-
-* The path to the output file
-    * The output file is a zip file containing the certificate signing requests
-      and private keys for each instance.
-* Information about each instance
-    * An instance is any piece of the Elastic Stack that requires a SSL certificate.
-      Depending on your configuration, Elasticsearch, Logstash, Kibana, and Beats
-      may all require a certificate and private key.
-    * The minimum required value for each instance is a name. This can simply be the
-      hostname, which will be used as the Common Name of the certificate. A full
-      distinguished name may also be used.
-    * IP addresses and DNS names are optional. Multiple values can be specified as a
-      comma separated string. If no IP addresses or DNS names are provided, you may
-      disable hostname verification in your SSL configuration.
-
-Let's get started...
-
-Please enter the desired output file [/home/es/config/x-pack/csr-bundle.zip]:
-Enter instance name: node01
-Enter name for directories and files [node01]:
-Enter IP Addresses for instance (comma-separated if more than one) []: 10.10.0.1
-Enter DNS names for instance (comma-separated if more than one) []: node01.mydomain.com,node01
-Would you like to specify another instance? Press 'y' to continue entering instance information: y
-Enter instance name: node02
-Enter name for directories and files [node02]:
-Enter IP Addresses for instance (comma-separated if more than one) []: 10.10.0.2
-Enter DNS names for instance (comma-separated if more than one) []: node02.mydomain.com
-Would you like to specify another instance? Press 'y' to continue entering instance information:
-Certificate signing requests written to /Users/jmodi/dev/tmp/elasticsearch-5.0.0-alpha5-SNAPSHOT/config/x-pack/csr-bundle.zip
-
-This file should be properly secured as it contains the private keys for all
-instances.
-
-After unzipping the file, there will be a directory for each instance containing
-the certificate signing request and the private key. Provide the certificate
-signing requests to your certificate authority. Once you have received the
-signed certificate, copy the signed certificate, key, and CA certificate to the
-configuration directory of the Elastic product that they will be used for and
-follow the SSL configuration instructions in the product guide.
-....
---------------------------------------------------
-
-In this case, the command generates a zip file with two CSRs and private
-keys. The CSRs should be provided to the CA in order to obtain the signed
-certificates. The signed certificates will need to be in PEM format in order to
-be used.
-////
 [float]
 [[certgen-silent]]
 ==== Using `certgen` in Silent Mode

docs/en/installing-xes.asciidoc

@@ -139,11 +139,17 @@ the correct value for your environment, you may consider setting the value to
 =============================================================================
 
 . Configure passwords and SSL/TLS.
-.. If you have more than one node or a single node that listens on an external
+.. If you have a non-trial license and you want to use {security}, you must
-interface, you must configure SSL/TLS for inter-node communication. Single-node
+configure SSL/TLS for internode-communication.
-instances that use a loopback interface do not have this requirement.  For more
++
+--
+NOTE: This requirement applies to clusters with more than one node and to
+clusters with a single node that listens on an external interface. Single-node
+clusters that use a loopback interface do not have this requirement.  For more 
 information, see
 {xpack-ref}/encrypting-communications.html[Encrypting Communications].
+
+--
 ... Generate node certificates. For example, you can use the `certgen` command
 line tool to generate a certificate authority and signed certificates for your
 nodes.

docs/en/security/authentication.asciidoc

@@ -154,11 +154,11 @@ be set before the user can be used.
 [[internal-users]]
 === Internal Users
 
-{security} has two _internal_ users (`_system` and `_xpack`) that are
+{security} has three _internal_ users (`_system`, `_xpack`, and `_xpack_security`)
-responsible for the operations that take place inside an Elasticsearch cluster.
+that are responsible for the operations that take place inside an {es} cluster.
 
 These users are only used by requests that originate from within the cluster.
-For this reason, they cannot be used to authenticate against the API, and there
+For this reason, they cannot be used to authenticate against the API and there
 is no password to manage or reset.
 
 From time-to-time you may find a reference to one of these users inside your

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

@@ -7,6 +7,9 @@ package org.elasticsearch.license;
 
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.master.MasterNodeReadRequest;
+import org.elasticsearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
 
 
 public class GetLicenseRequest extends MasterNodeReadRequest<GetLicenseRequest> {
@@ -14,6 +17,10 @@ public class GetLicenseRequest extends MasterNodeReadRequest<GetLicenseRequest>
     public GetLicenseRequest() {
     }
 
+    public GetLicenseRequest(StreamInput in) throws IOException {
+        super(in);
+    }
+
     @Override
     public ActionRequestValidationException validate() {
         return null;

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

@@ -27,8 +27,8 @@ public class TransportGetLicenseAction extends TransportMasterNodeReadAction<Get
     public TransportGetLicenseAction(Settings settings, TransportService transportService, ClusterService clusterService,
                                      LicenseService licenseService, ThreadPool threadPool, ActionFilters actionFilters,
                                      IndexNameExpressionResolver indexNameExpressionResolver) {
-        super(settings, GetLicenseAction.NAME, transportService, clusterService, threadPool, actionFilters, indexNameExpressionResolver,
+        super(settings, GetLicenseAction.NAME, transportService, clusterService, threadPool, actionFilters,
-                GetLicenseRequest::new);
+            GetLicenseRequest::new, indexNameExpressionResolver);
         this.licenseService = licenseService;
     }
 

plugin/src/main/java/org/elasticsearch/xpack/deprecation/DeprecationInfoAction.java

@@ -210,6 +210,17 @@ public class DeprecationInfoAction extends Action<DeprecationInfoAction.Request,
             this.indices = indices;
         }
 
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            indices = in.readStringArray();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeStringArray(indices);
+        }
+
         @Override
         public String[] indices() {
             return indices;
@@ -237,14 +248,7 @@ public class DeprecationInfoAction extends Action<DeprecationInfoAction.Request,
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
+            throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-            indices = in.readStringArray();
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeStringArray(indices);
         }
 
         @Override
@@ -286,7 +290,7 @@ public class DeprecationInfoAction extends Action<DeprecationInfoAction.Request,
                                IndexNameExpressionResolver indexNameExpressionResolver,
                                XPackLicenseState licenseState, InternalClient client) {
             super(settings, DeprecationInfoAction.NAME, transportService, clusterService, threadPool, actionFilters,
-                indexNameExpressionResolver, Request::new);
+                Request::new, indexNameExpressionResolver);
             this.licenseState = licenseState;
             this.client = client;
             this.indexNameExpressionResolver = indexNameExpressionResolver;

plugin/src/main/java/org/elasticsearch/xpack/ml/action/GetDatafeedsAction.java

@@ -79,6 +79,23 @@ public class GetDatafeedsAction extends Action<GetDatafeedsAction.Request, GetDa
             local(true);
         }
 
+        Request(StreamInput in) throws IOException {
+            super(in);
+            datafeedId = in.readString();
+            if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
+                allowNoDatafeeds = in.readBoolean();
+            }
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(datafeedId);
+            if (out.getVersion().onOrAfter(Version.V_6_1_0)) {
+                out.writeBoolean(allowNoDatafeeds);
+            }
+        }
+
         public String getDatafeedId() {
             return datafeedId;
         }
@@ -98,20 +115,7 @@ public class GetDatafeedsAction extends Action<GetDatafeedsAction.Request, GetDa
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
+            throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-            datafeedId = in.readString();
-            if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
-                allowNoDatafeeds = in.readBoolean();
-            }
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeString(datafeedId);
-            if (out.getVersion().onOrAfter(Version.V_6_1_0)) {
-                out.writeBoolean(allowNoDatafeeds);
-            }
         }
 
         @Override
@@ -202,7 +206,7 @@ public class GetDatafeedsAction extends Action<GetDatafeedsAction.Request, GetDa
         public TransportAction(Settings settings, TransportService transportService, ClusterService clusterService,
                 ThreadPool threadPool, ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver) {
             super(settings, GetDatafeedsAction.NAME, transportService, clusterService, threadPool, actionFilters,
-                    indexNameExpressionResolver, Request::new);
+                Request::new, indexNameExpressionResolver);
         }
 
         @Override

plugin/src/main/java/org/elasticsearch/xpack/ml/action/GetDatafeedsStatsAction.java

@@ -85,6 +85,23 @@ public class GetDatafeedsStatsAction extends Action<GetDatafeedsStatsAction.Requ
 
         Request() {}
 
+        Request(StreamInput in) throws IOException {
+            super(in);
+            datafeedId = in.readString();
+            if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
+                allowNoDatafeeds = in.readBoolean();
+            }
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(datafeedId);
+            if (out.getVersion().onOrAfter(Version.V_6_1_0)) {
+                out.writeBoolean(allowNoDatafeeds);
+            }
+        }
+
         public String getDatafeedId() {
             return datafeedId;
         }
@@ -104,20 +121,7 @@ public class GetDatafeedsStatsAction extends Action<GetDatafeedsStatsAction.Requ
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
+            throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-            datafeedId = in.readString();
-            if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
-                allowNoDatafeeds = in.readBoolean();
-            }
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeString(datafeedId);
-            if (out.getVersion().onOrAfter(Version.V_6_1_0)) {
-                out.writeBoolean(allowNoDatafeeds);
-            }
         }
 
         @Override
@@ -304,7 +308,7 @@ public class GetDatafeedsStatsAction extends Action<GetDatafeedsStatsAction.Requ
                                ThreadPool threadPool, ActionFilters actionFilters,
                                IndexNameExpressionResolver indexNameExpressionResolver) {
             super(settings, GetDatafeedsStatsAction.NAME, transportService, clusterService, threadPool, actionFilters,
-                    indexNameExpressionResolver, Request::new);
+                Request::new, indexNameExpressionResolver);
         }
 
         @Override

plugin/src/main/java/org/elasticsearch/xpack/ml/action/GetJobsAction.java

@@ -73,6 +73,23 @@ public class GetJobsAction extends Action<GetJobsAction.Request, GetJobsAction.R
             local(true);
         }
 
+        Request(StreamInput in) throws IOException {
+            super(in);
+            jobId = in.readString();
+            if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
+                allowNoJobs = in.readBoolean();
+            }
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(jobId);
+            if (out.getVersion().onOrAfter(Version.V_6_1_0)) {
+                out.writeBoolean(allowNoJobs);
+            }
+        }
+
         public void setAllowNoJobs(boolean allowNoJobs) {
             this.allowNoJobs = allowNoJobs;
         }
@@ -92,20 +109,7 @@ public class GetJobsAction extends Action<GetJobsAction.Request, GetJobsAction.R
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
+            throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-            jobId = in.readString();
-            if (in.getVersion().onOrAfter(Version.V_6_1_0)) {
-                allowNoJobs = in.readBoolean();
-            }
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeString(jobId);
-            if (out.getVersion().onOrAfter(Version.V_6_1_0)) {
-                out.writeBoolean(allowNoJobs);
-            }
         }
 
         @Override
@@ -197,10 +201,10 @@ public class GetJobsAction extends Action<GetJobsAction.Request, GetJobsAction.R
 
         @Inject
         public TransportAction(Settings settings, TransportService transportService, ClusterService clusterService,
-                ThreadPool threadPool, ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
+                               ThreadPool threadPool, ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
-                JobManager jobManager) {
+                               JobManager jobManager) {
             super(settings, GetJobsAction.NAME, transportService, clusterService, threadPool, actionFilters,
-                    indexNameExpressionResolver, Request::new);
+                Request::new, indexNameExpressionResolver);
             this.jobManager = jobManager;
         }
 

plugin/src/main/java/org/elasticsearch/xpack/security/authc/TokenService.java

@@ -221,8 +221,8 @@ public final class TokenService extends AbstractComponent {
 
     void decodeToken(String token, ActionListener<UserToken> listener) throws IOException {
         // We intentionally do not use try-with resources since we need to keep the stream open if we need to compute a key!
-        StreamInput in = new InputStreamStreamInput(
+        byte[] bytes = token.getBytes(StandardCharsets.UTF_8);
-                Base64.getDecoder().wrap(new ByteArrayInputStream(token.getBytes(StandardCharsets.UTF_8))));
+        StreamInput in = new InputStreamStreamInput(Base64.getDecoder().wrap(new ByteArrayInputStream(bytes)), bytes.length);
         if (in.available() < MINIMUM_BASE64_BYTES) {
             logger.debug("invalid token");
             listener.onResponse(null);

plugin/src/main/java/org/elasticsearch/xpack/upgrade/actions/IndexUpgradeAction.java

@@ -76,6 +76,17 @@ public class IndexUpgradeAction extends Action<IndexUpgradeAction.Request, BulkB
             this.index = index;
         }
 
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            index = in.readString();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(index);
+        }
+
         public String index() {
             return index;
         }
@@ -123,14 +134,7 @@ public class IndexUpgradeAction extends Action<IndexUpgradeAction.Request, BulkB
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
+            throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-            index = in.readString();
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeString(index);
         }
 
         @Override
@@ -179,7 +183,7 @@ public class IndexUpgradeAction extends Action<IndexUpgradeAction.Request, BulkB
                                IndexUpgradeService indexUpgradeService,
                                IndexNameExpressionResolver indexNameExpressionResolver) {
             super(settings, IndexUpgradeAction.NAME, transportService, clusterService, threadPool, actionFilters,
-                    indexNameExpressionResolver, Request::new);
+                Request::new, indexNameExpressionResolver);
             this.indexUpgradeService = indexUpgradeService;
         }
 

plugin/src/main/java/org/elasticsearch/xpack/upgrade/actions/IndexUpgradeInfoAction.java

@@ -135,6 +135,19 @@ public class IndexUpgradeInfoAction extends Action<IndexUpgradeInfoAction.Reques
             this.indices = indices;
         }
 
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            indices = in.readStringArray();
+            indicesOptions = IndicesOptions.readIndicesOptions(in);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeStringArray(indices);
+            indicesOptions.writeIndicesOptions(out);
+        }
+
         @Override
         public String[] indices() {
             return indices;
@@ -166,16 +179,7 @@ public class IndexUpgradeInfoAction extends Action<IndexUpgradeInfoAction.Reques
 
         @Override
         public void readFrom(StreamInput in) throws IOException {
-            super.readFrom(in);
+            throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-            indices = in.readStringArray();
-            indicesOptions = IndicesOptions.readIndicesOptions(in);
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeStringArray(indices);
-            indicesOptions.writeIndicesOptions(out);
         }
 
         @Override
@@ -223,7 +227,7 @@ public class IndexUpgradeInfoAction extends Action<IndexUpgradeInfoAction.Reques
                                IndexNameExpressionResolver indexNameExpressionResolver,
                                XPackLicenseState licenseState) {
             super(settings, IndexUpgradeInfoAction.NAME, transportService, clusterService, threadPool, actionFilters,
-                    indexNameExpressionResolver, Request::new);
+                Request::new, indexNameExpressionResolver);
             this.indexUpgradeService = indexUpgradeService;
             this.licenseState = licenseState;
         }

plugin/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/WatcherTransportAction.java

@@ -11,6 +11,7 @@ import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.license.XPackLicenseState;
@@ -19,8 +20,6 @@ import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.XPackPlugin;
 
-import java.util.function.Supplier;
-
 public abstract class WatcherTransportAction<Request extends ActionRequest, Response extends ActionResponse>
         extends HandledTransportAction<Request, Response> {
 
@@ -28,8 +27,8 @@ public abstract class WatcherTransportAction<Request extends ActionRequest, Resp
 
     public WatcherTransportAction(Settings settings, String actionName, TransportService transportService, ThreadPool threadPool,
                                   ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
-                                  XPackLicenseState licenseState, Supplier<Request> request) {
+                                  XPackLicenseState licenseState, Writeable.Reader<Request> request) {
-        super(settings, actionName, threadPool, transportService, actionFilters, indexNameExpressionResolver, request);
+        super(settings, actionName, threadPool, transportService, actionFilters, request, indexNameExpressionResolver);
         this.licenseState = licenseState;
     }
 

plugin/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/ack/AckWatchRequest.java

@@ -25,7 +25,7 @@ public class AckWatchRequest extends ActionRequest {
     private String[] actionIds = Strings.EMPTY_ARRAY;
 
     public AckWatchRequest() {
-        this(null);
+        this(null, (String[]) null);
     }
 
     public AckWatchRequest(String watchId, String... actionIds) {
@@ -33,6 +33,19 @@ public class AckWatchRequest extends ActionRequest {
         this.actionIds = actionIds;
     }
 
+    public AckWatchRequest(StreamInput in) throws IOException {
+        super(in);
+        watchId = in.readString();
+        actionIds = in.readStringArray();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeString(watchId);
+        out.writeStringArray(actionIds);
+    }
+
     /**
      * @return The id of the watch to be acked
      */
@@ -78,16 +91,7 @@ public class AckWatchRequest extends ActionRequest {
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-        watchId = in.readString();
-        actionIds = in.readStringArray();
-    }
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        super.writeTo(out);
-        out.writeString(watchId);
-        out.writeStringArray(actionIds);
     }
 
     @Override

plugin/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/activate/ActivateWatchRequest.java

@@ -31,6 +31,19 @@ public class ActivateWatchRequest extends ActionRequest {
         this.activate = activate;
     }
 
+    public ActivateWatchRequest(StreamInput in) throws IOException {
+        super(in);
+        watchId = in.readString();
+        activate = in.readBoolean();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeString(watchId);
+        out.writeBoolean(activate);
+    }
+
     /**
      * @return The id of the watch to be acked
      */
@@ -59,16 +72,7 @@ public class ActivateWatchRequest extends ActionRequest {
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-        watchId = in.readString();
-        activate = in.readBoolean();
-    }
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        super.writeTo(out);
-        out.writeString(watchId);
-        out.writeBoolean(activate);
     }
 
     @Override

plugin/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/execute/ExecuteWatchRequest.java

@@ -51,6 +51,56 @@ public class ExecuteWatchRequest extends ActionRequest {
         this.id = id;
     }
 
+    public ExecuteWatchRequest(StreamInput in) throws IOException {
+        super(in);
+        id = in.readOptionalString();
+        ignoreCondition = in.readBoolean();
+        recordExecution = in.readBoolean();
+        if (in.readBoolean()){
+            alternativeInput = in.readMap();
+        }
+        if (in.readBoolean()) {
+            triggerData = in.readMap();
+        }
+        long actionModesCount = in.readLong();
+        actionModes = new HashMap<>();
+        for (int i = 0; i < actionModesCount; i++) {
+            actionModes.put(in.readString(), ActionExecutionMode.resolve(in.readByte()));
+        }
+        if (in.readBoolean()) {
+            watchSource = in.readBytesReference();
+            xContentType = XContentType.readFrom(in);
+        }
+        debug = in.readBoolean();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeOptionalString(id);
+        out.writeBoolean(ignoreCondition);
+        out.writeBoolean(recordExecution);
+        out.writeBoolean(alternativeInput != null);
+        if (alternativeInput != null) {
+            out.writeMap(alternativeInput);
+        }
+        out.writeBoolean(triggerData != null);
+        if (triggerData != null) {
+            out.writeMap(triggerData);
+        }
+        out.writeLong(actionModes.size());
+        for (Map.Entry<String, ActionExecutionMode> entry : actionModes.entrySet()) {
+            out.writeString(entry.getKey());
+            out.writeByte(entry.getValue().id());
+        }
+        out.writeBoolean(watchSource != null);
+        if (watchSource != null) {
+            out.writeBytesReference(watchSource);
+            xContentType.writeTo(out);
+        }
+        out.writeBoolean(debug);
+    }
+
     /**
      * @return The id of the watch to be executed
      */
@@ -221,54 +271,7 @@ public class ExecuteWatchRequest extends ActionRequest {
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-        id = in.readOptionalString();
-        ignoreCondition = in.readBoolean();
-        recordExecution = in.readBoolean();
-        if (in.readBoolean()){
-            alternativeInput = in.readMap();
-        }
-        if (in.readBoolean()) {
-            triggerData = in.readMap();
-        }
-        long actionModesCount = in.readLong();
-        actionModes = new HashMap<>();
-        for (int i = 0; i < actionModesCount; i++) {
-            actionModes.put(in.readString(), ActionExecutionMode.resolve(in.readByte()));
-        }
-        if (in.readBoolean()) {
-            watchSource = in.readBytesReference();
-            xContentType = XContentType.readFrom(in);
-        }
-        debug = in.readBoolean();
-    }
-
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        super.writeTo(out);
-        out.writeOptionalString(id);
-        out.writeBoolean(ignoreCondition);
-        out.writeBoolean(recordExecution);
-        out.writeBoolean(alternativeInput != null);
-        if (alternativeInput != null) {
-            out.writeMap(alternativeInput);
-        }
-        out.writeBoolean(triggerData != null);
-        if (triggerData != null) {
-            out.writeMap(triggerData);
-        }
-        out.writeLong(actionModes.size());
-        for (Map.Entry<String, ActionExecutionMode> entry : actionModes.entrySet()) {
-            out.writeString(entry.getKey());
-            out.writeByte(entry.getValue().id());
-        }
-        out.writeBoolean(watchSource != null);
-        if (watchSource != null) {
-            out.writeBytesReference(watchSource);
-            xContentType.writeTo(out);
-        }
-        out.writeBoolean(debug);
     }
 
     @Override

plugin/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/get/GetWatchRequest.java

@@ -31,6 +31,17 @@ public class GetWatchRequest extends ActionRequest {
         this.id = id;
     }
 
+    public GetWatchRequest(StreamInput in) throws IOException {
+        super(in);
+        id = in.readString();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeString(id);
+    }
+
     GetWatchRequest setId(String id) {
         this.id = id;
         return this;
@@ -58,14 +69,7 @@ public class GetWatchRequest extends ActionRequest {
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-        id = in.readString();
-    }
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        super.writeTo(out);
-        out.writeString(id);
     }
 
     @Override

plugin/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/put/PutWatchRequest.java

@@ -42,6 +42,23 @@ public class PutWatchRequest extends ActionRequest {
         this.xContentType = xContentType;
     }
 
+    public PutWatchRequest(StreamInput in) throws IOException {
+        super(in);
+        id = in.readString();
+        source = in.readBytesReference();
+        active = in.readBoolean();
+        xContentType = XContentType.readFrom(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeString(id);
+        out.writeBytesReference(source);
+        out.writeBoolean(active);
+        xContentType.writeTo(out);
+    }
+
     /**
      * @return The name that will be the ID of the indexed document
      */
@@ -115,19 +132,6 @@ public class PutWatchRequest extends ActionRequest {
 
     @Override
     public void readFrom(StreamInput in) throws IOException {
-        super.readFrom(in);
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
-        id = in.readString();
-        source = in.readBytesReference();
-        active = in.readBoolean();
-        xContentType = XContentType.readFrom(in);
-    }
-
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        super.writeTo(out);
-        out.writeString(id);
-        out.writeBytesReference(source);
-        out.writeBoolean(active);
-        xContentType.writeTo(out);
     }
 }

plugin/src/main/resources/logstash-index-template.json

@@ -30,9 +30,6 @@
             }
           }
         },
-        "version":{
-          "type":"keyword"
-        },
         "pipeline":{
           "type":"text"
         },

plugin/src/test/java/org/elasticsearch/xpack/deprecation/DeprecationInfoActionRequestTests.java

@@ -5,9 +5,10 @@
  */
 package org.elasticsearch.xpack.deprecation;
 
-import org.elasticsearch.test.AbstractStreamableTestCase;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
 
-public class DeprecationInfoActionRequestTests extends AbstractStreamableTestCase<DeprecationInfoAction.Request> {
+public class DeprecationInfoActionRequestTests extends AbstractWireSerializingTestCase<DeprecationInfoAction.Request> {
 
     @Override
     protected DeprecationInfoAction.Request createTestInstance() {
@@ -15,7 +16,7 @@ public class DeprecationInfoActionRequestTests extends AbstractStreamableTestCas
     }
 
     @Override
-    protected DeprecationInfoAction.Request createBlankInstance() {
+    protected Writeable.Reader<DeprecationInfoAction.Request> instanceReader() {
-        return new DeprecationInfoAction.Request();
+        return DeprecationInfoAction.Request::new;
     }
 }

plugin/src/test/java/org/elasticsearch/xpack/ml/action/GetDatafeedStatsActionRequestTests.java

@@ -6,10 +6,11 @@
 package org.elasticsearch.xpack.ml.action;
 
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.test.AbstractStreamableTestCase;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.ml.action.GetDatafeedsStatsAction.Request;
 
-public class GetDatafeedStatsActionRequestTests extends AbstractStreamableTestCase<Request> {
+public class GetDatafeedStatsActionRequestTests extends AbstractWireSerializingTestCase<Request> {
 
     @Override
     protected Request createTestInstance() {
@@ -19,8 +20,7 @@ public class GetDatafeedStatsActionRequestTests extends AbstractStreamableTestCa
     }
 
     @Override
-    protected Request createBlankInstance() {
+    protected Writeable.Reader<Request> instanceReader() {
-        return new Request();
+        return Request::new;
     }
-
 }

plugin/src/test/java/org/elasticsearch/xpack/ml/action/GetDatafeedsActionRequestTests.java

@@ -6,10 +6,11 @@
 package org.elasticsearch.xpack.ml.action;
 
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.test.AbstractStreamableTestCase;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.ml.action.GetDatafeedsAction.Request;
 
-public class GetDatafeedsActionRequestTests extends AbstractStreamableTestCase<Request> {
+public class GetDatafeedsActionRequestTests extends AbstractWireSerializingTestCase<Request> {
 
     @Override
     protected Request createTestInstance() {
@@ -19,8 +20,7 @@ public class GetDatafeedsActionRequestTests extends AbstractStreamableTestCase<R
     }
 
     @Override
-    protected Request createBlankInstance() {
+    protected Writeable.Reader<Request> instanceReader() {
-        return new Request();
+        return Request::new;
     }
-
 }

plugin/src/test/java/org/elasticsearch/xpack/ml/action/GetJobsActionRequestTests.java

@@ -6,10 +6,11 @@
 package org.elasticsearch.xpack.ml.action;
 
 import org.elasticsearch.cluster.metadata.MetaData;
-import org.elasticsearch.test.AbstractStreamableTestCase;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.ml.action.GetJobsAction.Request;
 
-public class GetJobsActionRequestTests extends AbstractStreamableTestCase<GetJobsAction.Request> {
+public class GetJobsActionRequestTests extends AbstractWireSerializingTestCase<Request> {
 
     @Override
     protected Request createTestInstance() {
@@ -19,7 +20,7 @@ public class GetJobsActionRequestTests extends AbstractStreamableTestCase<GetJob
     }
 
     @Override
-    protected Request createBlankInstance() {
+    protected Writeable.Reader<Request> instanceReader() {
-        return new Request();
+        return Request::new;
     }
 }

plugin/src/test/java/org/elasticsearch/xpack/security/authc/AuthenticationServiceTests.java

@@ -17,6 +17,7 @@ import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.apache.lucene.util.ArrayUtil;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchSecurityException;
@@ -840,6 +841,11 @@ public class AuthenticationServiceTests extends ESTestCase {
                 assertEquals(expected, result);
                 latch.countDown();
             }, this::logAndFail));
+        } catch (IllegalArgumentException ex) {
+            assertThat(ex.getMessage(), containsString("array length must be <= to " + ArrayUtil.MAX_ARRAY_LENGTH  + " but was: "));
+        } catch (NegativeArraySizeException ex) {
+            assertThat(ex.getMessage(), containsString("array size must be positive but was: "));
+
         }
 
         // we need to use a latch here because the key computation goes async on another thread!

plugin/src/test/java/org/elasticsearch/xpack/upgrade/actions/IndexUpgradeActionRequestTests.java

@@ -5,17 +5,18 @@
  */
 package org.elasticsearch.xpack.upgrade.actions;
 
-import org.elasticsearch.test.AbstractStreamableTestCase;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.upgrade.actions.IndexUpgradeAction.Request;
 
-public class IndexUpgradeActionRequestTests extends AbstractStreamableTestCase<Request> {
+public class IndexUpgradeActionRequestTests extends AbstractWireSerializingTestCase<Request> {
     @Override
     protected Request createTestInstance() {
         return new Request(randomAlphaOfLength(10));
     }
 
     @Override
-    protected Request createBlankInstance() {
+    protected Writeable.Reader<Request> instanceReader() {
-        return new Request();
+        return Request::new;
     }
 }

plugin/src/test/java/org/elasticsearch/xpack/upgrade/actions/IndexUpgradeInfoActionRequestTests.java

@@ -6,10 +6,11 @@
 package org.elasticsearch.xpack.upgrade.actions;
 
 import org.elasticsearch.action.support.IndicesOptions;
-import org.elasticsearch.test.AbstractStreamableTestCase;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.upgrade.actions.IndexUpgradeInfoAction.Request;
 
-public class IndexUpgradeInfoActionRequestTests extends AbstractStreamableTestCase<Request> {
+public class IndexUpgradeInfoActionRequestTests extends AbstractWireSerializingTestCase<Request> {
     @Override
     protected Request createTestInstance() {
         int indexCount = randomInt(4);
@@ -25,7 +26,7 @@ public class IndexUpgradeInfoActionRequestTests extends AbstractStreamableTestCa
     }
 
     @Override
-    protected Request createBlankInstance() {
+    protected Writeable.Reader<Request> instanceReader() {
-        return new Request();
+        return Request::new;
     }
 }

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

@@ -93,7 +93,7 @@ public class WatchRequestValidationTests extends ESTestCase {
     }
 
     public void testGetWatchNullId() {
-        ActionRequestValidationException e = new GetWatchRequest(null).validate();
+        ActionRequestValidationException e = new GetWatchRequest((String) null).validate();
         assertThat(e, is(notNullValue()));
         assertThat(e.validationErrors(), hasItem("watch id is missing"));
     }
@@ -105,7 +105,7 @@ public class WatchRequestValidationTests extends ESTestCase {
     }
 
     public void testExecuteWatchMissingWatchIdNoSource() {
-        ActionRequestValidationException e = new ExecuteWatchRequest(null).validate();
+        ActionRequestValidationException e = new ExecuteWatchRequest((String) null).validate();
         assertThat(e, is(notNullValue()));
         assertThat(e.validationErrors(),
                 hasItem("a watch execution request must either have a watch id or an inline watch source, but both are missing"));

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

@@ -25,8 +25,7 @@ public class ExecuteWatchRequestTests extends ESTestCase {
         BytesStreamOutput out = new BytesStreamOutput();
         request.writeTo(out);
         StreamInput in = StreamInput.wrap(out.bytes().toBytesRef().bytes);
-        ExecuteWatchRequest serialized = new ExecuteWatchRequest();
+        ExecuteWatchRequest serialized = new ExecuteWatchRequest(in);
-        serialized.readFrom(in);
         assertEquals(XContentType.JSON, serialized.getXContentType());
         assertEquals("{}", serialized.getWatchSource().utf8ToString());
     }

plugin/src/test/java/org/elasticsearch/xpack/watcher/transport/action/put/PutWatchSerializationTests.java

@@ -29,8 +29,7 @@ public class PutWatchSerializationTests extends ESTestCase {
         BytesStreamOutput streamOutput = new BytesStreamOutput();
         request.writeTo(streamOutput);
 
-        PutWatchRequest readRequest = new PutWatchRequest();
+        PutWatchRequest readRequest = new PutWatchRequest(streamOutput.bytes().streamInput());
-        readRequest.readFrom(streamOutput.bytes().streamInput());
         assertThat(readRequest.isActive(), is(request.isActive()));
         assertThat(readRequest.getId(), is(request.getId()));
         assertThat(readRequest.getSource(), is(request.getSource()));
@@ -49,9 +48,7 @@ public class PutWatchSerializationTests extends ESTestCase {
         BytesStreamOutput streamOutput = new BytesStreamOutput();
         request.writeTo(streamOutput);
 
-        PutWatchRequest readRequest = new PutWatchRequest();
+        PutWatchRequest readRequest = new PutWatchRequest(streamOutput.bytes().streamInput());
-        StreamInput input = streamOutput.bytes().streamInput();
-        readRequest.readFrom(input);
         assertThat(readRequest.isActive(), is(request.isActive()));
         assertThat(readRequest.getId(), is(request.getId()));
         assertThat(readRequest.getSource(), is(request.getSource()));

plugin/src/test/resources/wrong-version-logstash-index-template.json

@@ -30,9 +30,6 @@
             }
           }
         },
-        "version":{
-          "type":"keyword"
-        },
         "pipeline":{
           "type":"text"
         },