Fix search scroll request with a plain text body (#23183)

The backport of #22691 caused plain text bodies with a scroll id to fail with an IllegalStateException as the wrong
method was being called. This commit adds tests to ensure plain text bodies work and fixes the search scroll action so
that it properly handles a request with a plain text body.

core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java

@@ -57,7 +57,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
         }
 
-        request.withContentOrSourceParamParserOrNull(xContentParser -> {
+        request.withContentOrSourceParamParserOrNullLenient(xContentParser -> {
             if (xContentParser == null) {
                 if (request.hasContent()) {
                     // TODO: why do we accept this plain text value? maybe we can just use the scroll params?

core/src/test/java/org/elasticsearch/search/scroll/RestClearScrollActionTests.java

@@ -19,21 +19,33 @@
 
 package org.elasticsearch.search.scroll;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.ClearScrollRequest;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.search.RestClearScrollAction;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.rest.FakeRestRequest;
+import org.mockito.ArgumentCaptor;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.elasticsearch.mock.orig.Mockito.verify;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.startsWith;
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 
 public class RestClearScrollActionTests extends ESTestCase {
@@ -68,4 +80,19 @@ public void testParseClearScrollRequestWithUnknownParamThrowsException() throws
         assertThat(e.getMessage(), startsWith("Unknown parameter [unknown]"));
     }
 
+    public void testParseClearScrollPlaintext() throws Exception {
+        RestClearScrollAction action = new RestClearScrollAction(Settings.EMPTY, mock(RestController.class));
+        NodeClient mockNodeClient = mock(NodeClient.class);
+        final List<String> scrollIds = Arrays.asList(generateRandomStringArray(4, 30, false, false));
+        final String content = scrollIds.stream().collect(Collectors.joining(","));
+        FakeRestRequest fakeRestRequest = new FakeRestRequest.Builder(NamedXContentRegistry.EMPTY)
+            .withContent(new BytesArray(content), null)
+            .withHeaders(Collections.singletonMap("Content-Type", Collections.singletonList("text/plain")))
+            .build();
+        action.handleRequest(fakeRestRequest, mock(RestChannel.class), mockNodeClient);
+        ArgumentCaptor<ClearScrollRequest> captor = ArgumentCaptor.forClass(ClearScrollRequest.class);
+        verify(mockNodeClient).clearScroll(captor.capture(), any(ActionListener.class));
+
+        assertEquals(scrollIds, captor.getValue().getScrollIds());
+    }
 }

core/src/test/java/org/elasticsearch/search/scroll/RestSearchScrollActionTests.java

@@ -19,21 +19,30 @@
 
 package org.elasticsearch.search.scroll;
 
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchScrollRequest;
+import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.search.RestSearchScrollAction;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.rest.FakeRestRequest;
+import org.mockito.ArgumentCaptor;
 
+import java.util.Collections;
+
+import static org.elasticsearch.mock.orig.Mockito.verify;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.startsWith;
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 
 public class RestSearchScrollActionTests extends ESTestCase {
@@ -71,4 +80,20 @@ public void testParseSearchScrollRequestWithUnknownParamThrowsException() throws
                 () -> RestSearchScrollAction.buildFromContent(invalidContent, searchScrollRequest));
         assertThat(e.getMessage(), startsWith("Unknown parameter [unknown]"));
     }
+
+    public void testParseSearchScrollPlaintext() throws Exception {
+        RestSearchScrollAction action = new RestSearchScrollAction(Settings.EMPTY, mock(RestController.class));
+        NodeClient mockNodeClient = mock(NodeClient.class);
+        final String scrollId = randomAsciiOfLengthBetween(1, 30);
+        FakeRestRequest fakeRestRequest = new FakeRestRequest.Builder(NamedXContentRegistry.EMPTY)
+            .withContent(new BytesArray(scrollId), null)
+            .withHeaders(Collections.singletonMap("Content-Type", Collections.singletonList("text/plain")))
+            .build();
+        action.handleRequest(fakeRestRequest, mock(RestChannel.class), mockNodeClient);
+        ArgumentCaptor<SearchScrollRequest> captor = ArgumentCaptor.forClass(SearchScrollRequest.class);
+        verify(mockNodeClient).searchScroll(captor.capture(), any(ActionListener.class));
+
+        assertEquals(scrollId, captor.getValue().scrollId());
+    }
+
 }