[SEARCH] Passing fieddata_fields as a non array causes OOM

If `fielddata_fields` are passed as a simple value instead of an array
we end up in an infinite loop createing parsed elements with null
values.
This commit validates the incoming token

Closes #8203

src/main/java/org/elasticsearch/search/fetch/fielddata/FieldDataFieldsParseElement.java

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.search.fetch.fielddata;
 
+import org.elasticsearch.ElasticsearchIllegalStateException;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.search.SearchParseElement;
 import org.elasticsearch.search.internal.SearchContext;
@@ -36,10 +37,17 @@ import org.elasticsearch.search.internal.SearchContext;
 public class FieldDataFieldsParseElement implements SearchParseElement {
     @Override
     public void parse(XContentParser parser, SearchContext context) throws Exception {
-        XContentParser.Token token;
+        XContentParser.Token token = parser.currentToken();
-        while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
+        if (token == XContentParser.Token.START_ARRAY) {
+            while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
                 String fieldName = parser.text();
                 context.fieldDataFields().add(new FieldDataFieldsContext.FieldDataField(fieldName));
             }
+        } else if (token == XContentParser.Token.VALUE_STRING) {
+            String fieldName = parser.text();
+            context.fieldDataFields().add(new FieldDataFieldsContext.FieldDataField(fieldName));
+        }  else {
+            throw new ElasticsearchIllegalStateException("Expected either a VALUE_STRING or an START_ARRAY but got " + token);
+        }
     }
 }

src/test/java/org/elasticsearch/search/fields/SearchFieldsTests.java

@@ -19,7 +19,9 @@
 
 package org.elasticsearch.search.fields;
 
+import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.search.SearchPhaseExecutionException;
 import org.elasticsearch.action.search.SearchRequestBuilder;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.common.Base64;
@@ -31,6 +33,7 @@ import org.elasticsearch.common.joda.Joda;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.search.SearchHitField;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
 import org.joda.time.DateTime;
@@ -40,11 +43,13 @@ import org.junit.Test;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
 
 import static org.elasticsearch.client.Requests.refreshRequest;
 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertFailures;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.hamcrest.Matchers.*;
 
@@ -476,6 +481,27 @@ public class SearchFieldsTests extends ElasticsearchIntegrationTest {
         assertThat(searchResponse.getHits().getAt(0).field(field).getValues().get(1).toString(), equalTo("value2"));
     }
 
+    @Test // see #8203
+    public void testSingleValueFieldDatatField() throws ExecutionException, InterruptedException {
+        createIndex("test");
+        indexRandom(true, client().prepareIndex("test", "type", "1").setSource("test_field", "foobar"));
+        refresh();
+        SearchResponse searchResponse = client().prepareSearch("test").setTypes("type").setSource(new BytesArray(new BytesRef("{\"query\":{\"match_all\":{}},\"fielddata_fields\": \"test_field\"}"))).get();
+        assertHitCount(searchResponse, 1);
+        Map<String,SearchHitField> fields = searchResponse.getHits().getHits()[0].getFields();
+        assertThat((String)fields.get("test_field").value(), equalTo("foobar"));
+    }
+
+    @Test(expected = SearchPhaseExecutionException.class)
+    public void testInvalidFieldDataField() throws ExecutionException, InterruptedException {
+        createIndex("test");
+        if (randomBoolean()) {
+            client().prepareSearch("test").setTypes("type").setSource(new BytesArray(new BytesRef("{\"query\":{\"match_all\":{}},\"fielddata_fields\": {}}"))).get();
+        } else {
+            client().prepareSearch("test").setTypes("type").setSource(new BytesArray(new BytesRef("{\"query\":{\"match_all\":{}},\"fielddata_fields\": 1.0}"))).get();
+        }
+    }
+
     @Test
     public void testFieldsPulledFromFieldData() throws Exception {
         createIndex("test");