Merge remote-tracking branch 'origin/master' into feature/client_aggs_parsing

core/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java

@@ -91,7 +91,13 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor {
         for (Runnable runnable : runnables) {
             if (runnable instanceof TieBreakingPrioritizedRunnable) {
                 TieBreakingPrioritizedRunnable t = (TieBreakingPrioritizedRunnable) runnable;
-                pending.add(new Pending(unwrap(t.runnable), t.priority(), t.insertionOrder, executing));
+                Runnable innerRunnable = t.runnable;
+                if (innerRunnable != null) {
+                    /** innerRunnable can be null if task is finished but not removed from executor yet,
+                     * see {@link TieBreakingPrioritizedRunnable#run} and {@link TieBreakingPrioritizedRunnable#runAndClean}
+                     */
+                    pending.add(new Pending(unwrap(innerRunnable), t.priority(), t.insertionOrder, executing));
+                }
             } else if (runnable instanceof PrioritizedFutureTask) {
                 PrioritizedFutureTask t = (PrioritizedFutureTask) runnable;
                 Object task = t.task;

core/src/main/java/org/elasticsearch/common/xcontent/XContentParserUtils.java

@@ -20,8 +20,10 @@
 package org.elasticsearch.common.xcontent;
 
 import org.elasticsearch.common.ParsingException;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.xcontent.XContentParser.Token;
+import org.elasticsearch.rest.action.search.RestSearchAction;
 
 import java.io.IOException;
 import java.util.Locale;
@@ -107,4 +109,37 @@ public final class XContentParserUtils {
         }
         return value;
     }
+
+    /**
+     * This method expects that the current token is a {@code XContentParser.Token.FIELD_NAME} and
+     * that the current field name is the concatenation of a type, delimiter and name (ex: terms#foo
+     * where "terms" refers to the type of a registered {@link NamedXContentRegistry.Entry}, "#" is
+     * the delimiter and "foo" the name of the object to parse).
+     *
+     * The method splits the field's name to extract the type and name and then parses the object
+     * using the {@link XContentParser#namedObject(Class, String, Object)} method.
+     *
+     * @param parser      the current {@link XContentParser}
+     * @param delimiter   the delimiter to use to splits the field's name
+     * @param objectClass the object class of the object to parse
+     * @param <T>         the type of the object to parse
+     * @return the parsed object
+     * @throws IOException if anything went wrong during parsing or if the type or name cannot be derived
+     *                     from the field's name
+     */
+    public static <T> T parseTypedKeysObject(XContentParser parser, String delimiter, Class<T> objectClass) throws IOException {
+        ensureExpectedToken(XContentParser.Token.FIELD_NAME, parser.currentToken(), parser::getTokenLocation);
+        String currentFieldName = parser.currentName();
+        if (Strings.hasLength(currentFieldName)) {
+            int position = currentFieldName.indexOf(delimiter);
+            if (position > 0) {
+                String type = currentFieldName.substring(0, position);
+                String name = currentFieldName.substring(position + 1);
+                return parser.namedObject(objectClass, type, name);
+            }
+        }
+        throw new ParsingException(parser.getTokenLocation(), "Cannot parse object of class [" + objectClass.getSimpleName()
+                + "] without type information. Set [" + RestSearchAction.TYPED_KEYS_PARAM + "] parameter on the request to ensure the"
+                + " type information is added to the response output");
+    }
 }

core/src/main/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregation.java

@@ -20,6 +20,7 @@
 package org.elasticsearch.search.aggregations;
 
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.search.aggregations.bucket.MultiBucketsAggregation;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 
@@ -81,7 +82,7 @@ public abstract class InternalMultiBucketAggregation<A extends InternalMultiBuck
         }
     }
 
-    public abstract static class InternalBucket implements Bucket {
+    public abstract static class InternalBucket implements Bucket, Writeable {
 
         public Object getProperty(String containingAggName, List<String> path) {
             if (path.isEmpty()) {

core/src/main/java/org/elasticsearch/search/aggregations/bucket/MultiBucketsAggregation.java

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.search.aggregations.bucket;
 
-import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.Comparators;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.search.aggregations.Aggregation;
@@ -37,7 +36,7 @@ public interface MultiBucketsAggregation extends Aggregation {
      * A bucket represents a criteria to which all documents that fall in it adhere to. It is also uniquely identified
      * by a key, and can potentially hold sub-aggregations computed over all documents in it.
      */
-    interface Bucket extends HasAggregations, ToXContent, Writeable {
+    interface Bucket extends HasAggregations, ToXContent {
         /**
          * @return The key associated with the bucket
          */

core/src/main/java/org/elasticsearch/search/suggest/Suggest.java

@@ -21,7 +21,6 @@ package org.elasticsearch.search.suggest;
 import org.apache.lucene.util.CollectionUtil;
 import org.elasticsearch.common.CheckedFunction;
 import org.elasticsearch.common.ParseField;
-import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Streamable;
@@ -33,6 +32,7 @@ import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentParserUtils;
 import org.elasticsearch.rest.action.search.RestSearchAction;
 import org.elasticsearch.search.aggregations.Aggregation;
 import org.elasticsearch.search.suggest.Suggest.Suggestion.Entry;
@@ -386,22 +386,7 @@ public class Suggest implements Iterable<Suggest.Suggestion<? extends Entry<? ex
 
         @SuppressWarnings("unchecked")
         public static Suggestion<? extends Entry<? extends Option>> fromXContent(XContentParser parser) throws IOException {
-            ensureExpectedToken(XContentParser.Token.FIELD_NAME, parser.currentToken(), parser::getTokenLocation);
+            return XContentParserUtils.parseTypedKeysObject(parser, Aggregation.TYPED_KEYS_DELIMITER, Suggestion.class);
-            String typeAndName = parser.currentName();
-            // we need to extract the type prefix from the name and throw error if it is not present
-            int delimiterPos = typeAndName.indexOf(Aggregation.TYPED_KEYS_DELIMITER);
-            String type;
-            String name;
-            if (delimiterPos > 0) {
-                type = typeAndName.substring(0, delimiterPos);
-                name = typeAndName.substring(delimiterPos + 1);
-            } else {
-                throw new ParsingException(parser.getTokenLocation(),
-                        "Cannot parse suggestion response without type information. Set [" + RestSearchAction.TYPED_KEYS_PARAM
-                                + "] parameter on the request to ensure the type information is added to the response output");
-            }
-
-            return parser.namedObject(Suggestion.class, type, name);
         }
 
         protected static <E extends Suggestion.Entry<?>> void parseEntries(XContentParser parser, Suggestion<E> suggestion,

core/src/test/java/org/elasticsearch/common/xcontent/XContentParserUtilsTests.java

@@ -19,15 +19,22 @@
 
 package org.elasticsearch.common.xcontent;
 
+import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.ParsingException;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.ESTestCase;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 
+import static org.elasticsearch.common.xcontent.XContentHelper.toXContent;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
+import static org.elasticsearch.common.xcontent.XContentParserUtils.parseTypedKeysObject;
 
 public class XContentParserUtilsTests extends ESTestCase {
+
     public void testEnsureExpectedToken() throws IOException {
         final XContentParser.Token randomToken = randomFrom(XContentParser.Token.values());
         try (XContentParser parser = createParser(JsonXContent.jsonXContent, "{}")) {
@@ -40,4 +47,68 @@ public class XContentParserUtilsTests extends ESTestCase {
             ensureExpectedToken(XContentParser.Token.END_OBJECT, parser.nextToken(), parser::getTokenLocation);
         }
     }
+
+    public void testParseTypedKeysObject() throws IOException {
+        final String delimiter = randomFrom("#", ":", "/", "-", "_", "|", "_delim_");
+        final XContentType xContentType = randomFrom(XContentType.values());
+
+        List<NamedXContentRegistry.Entry> namedXContents = new ArrayList<>();
+        namedXContents.add(new NamedXContentRegistry.Entry(Boolean.class, new ParseField("bool"), parser -> {
+            ensureExpectedToken(XContentParser.Token.VALUE_BOOLEAN, parser.nextToken(), parser::getTokenLocation);
+            return parser.booleanValue();
+        }));
+        namedXContents.add(new NamedXContentRegistry.Entry(Long.class, new ParseField("long"), parser -> {
+            ensureExpectedToken(XContentParser.Token.VALUE_NUMBER, parser.nextToken(), parser::getTokenLocation);
+            return parser.longValue();
+        }));
+        final NamedXContentRegistry namedXContentRegistry = new NamedXContentRegistry(namedXContents);
+
+        BytesReference bytes = toXContent((builder, params) -> builder.field("test", 0), xContentType, randomBoolean());
+        try (XContentParser parser = xContentType.xContent().createParser(namedXContentRegistry, bytes)) {
+            parser.nextToken();
+            ParsingException e = expectThrows(ParsingException.class, () -> parseTypedKeysObject(parser, delimiter, Boolean.class));
+            assertEquals("Failed to parse object: expecting token of type [FIELD_NAME] but found [START_OBJECT]", e.getMessage());
+
+            parser.nextToken();
+            e = expectThrows(ParsingException.class, () -> parseTypedKeysObject(parser, delimiter, Boolean.class));
+            assertEquals("Cannot parse object of class [Boolean] without type information. Set [typed_keys] parameter " +
+                    "on the request to ensure the type information is added to the response output", e.getMessage());
+        }
+
+        bytes = toXContent((builder, params) -> builder.field("type" + delimiter + "name", 0), xContentType, randomBoolean());
+        try (XContentParser parser = xContentType.xContent().createParser(namedXContentRegistry, bytes)) {
+            ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser::getTokenLocation);
+            ensureExpectedToken(XContentParser.Token.FIELD_NAME, parser.nextToken(), parser::getTokenLocation);
+
+            NamedXContentRegistry.UnknownNamedObjectException e = expectThrows(NamedXContentRegistry.UnknownNamedObjectException.class,
+                    () -> parseTypedKeysObject(parser, delimiter, Boolean.class));
+            assertEquals("Unknown Boolean [type]", e.getMessage());
+            assertEquals("type", e.getName());
+            assertEquals("java.lang.Boolean", e.getCategoryClass());
+        }
+
+        final long longValue = randomLong();
+        final boolean boolValue = randomBoolean();
+        bytes = toXContent((builder, params) -> {
+            builder.field("long" + delimiter + "l", longValue);
+            builder.field("bool" + delimiter + "b", boolValue);
+            return builder;
+        }, xContentType, randomBoolean());
+
+        try (XContentParser parser = xContentType.xContent().createParser(namedXContentRegistry, bytes)) {
+            ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser::getTokenLocation);
+
+            ensureExpectedToken(XContentParser.Token.FIELD_NAME, parser.nextToken(), parser::getTokenLocation);
+            Long parsedLong = parseTypedKeysObject(parser, delimiter, Long.class);
+            assertNotNull(parsedLong);
+            assertEquals(longValue, parsedLong.longValue());
+
+            ensureExpectedToken(XContentParser.Token.FIELD_NAME, parser.nextToken(), parser::getTokenLocation);
+            Boolean parsedBoolean = parseTypedKeysObject(parser, delimiter, Boolean.class);
+            assertNotNull(parsedBoolean);
+            assertEquals(boolValue, parsedBoolean);
+
+            ensureExpectedToken(XContentParser.Token.END_OBJECT, parser.nextToken(), parser::getTokenLocation);
+        }
+    }
 }

core/src/test/java/org/elasticsearch/search/suggest/SuggestionTests.java

@@ -133,7 +133,7 @@ public class SuggestionTests extends ESTestCase {
             ensureExpectedToken(XContentParser.Token.FIELD_NAME, parser.nextToken(), parser::getTokenLocation);
             ParsingException e = expectThrows(ParsingException.class, () -> Suggestion.fromXContent(parser));
             assertEquals(
-                    "Cannot parse suggestion response without type information. "
+                    "Cannot parse object of class [Suggestion] without type information. "
                     + "Set [typed_keys] parameter on the request to ensure the type information "
                     + "is added to the response output", e.getMessage());
         }