Merge pull request #16861 from nik9000/reindex_is_ready

Reindex required some parsing changes for search requests to support
differing defaults from the regular search api.

Author: nik9000

core/src/main/java/org/elasticsearch/action/ActionListener.java

@@ -21,18 +21,16 @@
 
 /**
  * A listener for action responses or failures.
- *
- *
  */
 public interface ActionListener<Response> {
-
     /**
-     * A response handler.
+     * Handle action response. This response may constitute a failure or a
+     * success but it is up to the listener to make that decision.
      */
     void onResponse(Response response);
 
     /**
-     * A failure handler.
+     * A failure caused by an exception at some phase of the task.
      */
     void onFailure(Throwable e);
 }

core/src/main/java/org/elasticsearch/action/bulk/BulkItemResponse.java

@@ -28,7 +28,9 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Streamable;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.StatusToXContent;
+import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentBuilderString;
 import org.elasticsearch.rest.RestStatus;
@@ -76,7 +78,15 @@ static final class Fields {
     /**
      * Represents a failure.
      */
-    public static class Failure {
+    public static class Failure implements Writeable<Failure>, ToXContent {
+        static final String INDEX_FIELD = "index";
+        static final String TYPE_FIELD = "type";
+        static final String ID_FIELD = "id";
+        static final String CAUSE_FIELD = "cause";
+        static final String STATUS_FIELD = "status";
+
+        public static final Failure PROTOTYPE = new Failure(null, null, null, null);
+
         private final String index;
         private final String type;
         private final String id;
@@ -126,9 +136,39 @@ public RestStatus getStatus() {
             return this.status;
         }
 
+        /**
+         * The actual cause of the failure.
+         */
         public Throwable getCause() {
             return cause;
         }
+
+        @Override
+        public Failure readFrom(StreamInput in) throws IOException {
+            return new Failure(in.readString(), in.readString(), in.readOptionalString(), in.readThrowable());
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeString(getIndex());
+            out.writeString(getType());
+            out.writeOptionalString(getId());
+            out.writeThrowable(getCause());
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.field(INDEX_FIELD, index);
+            builder.field(TYPE_FIELD, type);
+            if (id != null) {
+                builder.field(ID_FIELD, id);
+            }
+            builder.startObject(CAUSE_FIELD);
+            ElasticsearchException.toXContent(builder, params, cause);
+            builder.endObject();
+            builder.field(STATUS_FIELD, status.getStatus());
+            return builder;
+        }
     }
 
     private int id;
@@ -265,11 +305,7 @@ public void readFrom(StreamInput in) throws IOException {
         }
 
         if (in.readBoolean()) {
-            String fIndex = in.readString();
-            String fType = in.readString();
-            String fId = in.readOptionalString();
-            Throwable throwable = in.readThrowable();
-            failure = new Failure(fIndex, fType, fId, throwable);
+            failure = Failure.PROTOTYPE.readFrom(in);
         }
     }
 
@@ -294,10 +330,7 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeBoolean(false);
         } else {
             out.writeBoolean(true);
-            out.writeString(failure.getIndex());
-            out.writeString(failure.getType());
-            out.writeOptionalString(failure.getId());
-            out.writeThrowable(failure.getCause());
+            failure.writeTo(out);
         }
     }
 }

core/src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java

@@ -94,6 +94,12 @@ public void readFrom(StreamInput in) throws IOException {
 
     @Override
     public String toString() {
-        return "shard bulk {" + super.toString() + "}";
+        // This is included in error messages so we'll try to make it somewhat user friendly.
+        StringBuilder b = new StringBuilder("BulkShardRequest to [");
+        b.append(index).append("] containing [").append(items.length).append("] requests");
+        if (refresh) {
+            b.append(" and a refresh");
+        }
+        return b.toString();
     }
 }

core/src/main/java/org/elasticsearch/action/bulk/Retry.java

@@ -38,7 +38,7 @@
 /**
  * Encapsulates synchronous and asynchronous retry logic.
  */
-class Retry {
+public class Retry {
     private final Class<? extends Throwable> retryOnThrowable;
 
     private BackoffPolicy backoffPolicy;

core/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -223,6 +223,13 @@ public ActionRequestValidationException validate() {
         return validationException;
     }
 
+    /**
+     * The content type that will be used when generating a document from user provided objects like Maps.
+     */
+    public XContentType getContentType() {
+        return contentType;
+    }
+
     /**
      * Sets the content type that will be used when generating a document from user provided objects (like Map).
      */
@@ -294,6 +301,7 @@ public IndexRequest parent(String parent) {
         return this;
     }
 
+    @Override
     public String parent() {
         return this.parent;
     }
@@ -645,7 +653,7 @@ public void process(MetaData metaData, @Nullable MappingMetaData mappingMd, bool
     @Override
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        type = in.readString();
+        type = in.readOptionalString();
         id = in.readOptionalString();
         routing = in.readOptionalString();
         parent = in.readOptionalString();
@@ -663,7 +671,7 @@ public void readFrom(StreamInput in) throws IOException {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeString(type);
+        out.writeOptionalString(type);
         out.writeOptionalString(id);
         out.writeOptionalString(routing);
         out.writeOptionalString(parent);

core/src/main/java/org/elasticsearch/action/support/TransportAction.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskListener;
 import org.elasticsearch.tasks.TaskManager;
 import org.elasticsearch.threadpool.ThreadPool;
 
@@ -72,6 +73,13 @@ public final ActionFuture<Response> execute(Request request) {
      * This is a typical behavior.
      */
     public final Task execute(Request request, ActionListener<Response> listener) {
+        /*
+         * While this version of execute could delegate to the TaskListener
+         * version of execute that'd add yet another layer of wrapping on the
+         * listener and prevent us from using the listener bare if there isn't a
+         * task. That just seems like too many objects. Thus the two versions of
+         * this method.
+         */
         Task task = taskManager.register("transport", actionName, request);
         if (task == null) {
             execute(null, request, listener);
@@ -93,11 +101,32 @@ public void onFailure(Throwable e) {
         return task;
     }
 
+    public final Task execute(Request request, TaskListener<Response> listener) {
+        Task task = taskManager.register("transport", actionName, request);
+        execute(task, request, new ActionListener<Response>() {
+            @Override
+            public void onResponse(Response response) {
+                if (task != null) {
+                    taskManager.unregister(task);
+                }
+                listener.onResponse(task, response);
+            }
+
+            @Override
+            public void onFailure(Throwable e) {
+                if (task != null) {
+                    taskManager.unregister(task);
+                }
+                listener.onFailure(task, e);
+            }
+        });
+        return task;
+    }
+
     /**
      * Use this method when the transport action should continue to run in the context of the current task
      */
     public final void execute(Task task, Request request, ActionListener<Response> listener) {
-
         ActionRequestValidationException validationException = request.validate();
         if (validationException != null) {
             listener.onFailure(validationException);