Wrap stream passed to createParser in try-with-resources (#28897)

* Wrap stream passed to createParser in try-with-resources

This wraps the stream (`.streamInput()`) that is passed to many of the
`createParser` instances in the enclosing (or a new) try-with-resources block.
This ensures the `BytesReference.streamInput()` is closed.

Relates to #28504

* Use try-with-resources instead of closing in a finally block

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/JsonProcessor.java

@@ -38,6 +38,7 @@ import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.Map;
 
 import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
@@ -77,8 +78,9 @@ public final class JsonProcessor extends AbstractProcessor {
     public void execute(IngestDocument document) throws Exception {
         Object fieldValue = document.getFieldValue(field, Object.class);
         BytesReference bytesRef = (fieldValue == null) ? new BytesArray("null") : new BytesArray(fieldValue.toString());
-        try (XContentParser parser = JsonXContent.jsonXContent
-                .createParser(NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION, bytesRef.streamInput())) {
+        try (InputStream stream = bytesRef.streamInput();
+             XContentParser parser = JsonXContent.jsonXContent
+                .createParser(NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION, stream)) {
             XContentParser.Token token = parser.nextToken();
             Object value = null;
             if (token == XContentParser.Token.VALUE_NULL) {

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ScriptProcessor.java

@@ -36,6 +36,7 @@ import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptException;
 import org.elasticsearch.script.ScriptService;
 
+import java.io.InputStream;
 import java.util.Arrays;
 import java.util.Map;
 
@@ -97,21 +98,23 @@ public final class ScriptProcessor extends AbstractProcessor {
         @Override
         public ScriptProcessor create(Map<String, Processor.Factory> registry, String processorTag,
                                       Map<String, Object> config) throws Exception {
-            XContentBuilder builder = XContentBuilder.builder(JsonXContent.jsonXContent).map(config);
-            XContentParser parser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
-                LoggingDeprecationHandler.INSTANCE, builder.bytes().streamInput());
-            Script script = Script.parse(parser);
-
-            Arrays.asList("id", "source", "inline", "lang", "params", "options").forEach(config::remove);
-
-            // verify script is able to be compiled before successfully creating processor.
-            try {
-                scriptService.compile(script, ExecutableScript.INGEST_CONTEXT);
-            } catch (ScriptException e) {
-                throw newConfigurationException(TYPE, processorTag, null, e);
+            try (XContentBuilder builder = XContentBuilder.builder(JsonXContent.jsonXContent).map(config);
+                 InputStream stream = builder.bytes().streamInput();
+                 XContentParser parser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
+                     LoggingDeprecationHandler.INSTANCE, stream)) {
+                Script script = Script.parse(parser);
+
+                Arrays.asList("id", "source", "inline", "lang", "params", "options").forEach(config::remove);
+
+                // verify script is able to be compiled before successfully creating processor.
+                try {
+                    scriptService.compile(script, ExecutableScript.INGEST_CONTEXT);
+                } catch (ScriptException e) {
+                    throw newConfigurationException(TYPE, processorTag, null, e);
+                }
+
+                return new ScriptProcessor(processorTag, script, scriptService);
             }
-
-            return new ScriptProcessor(processorTag, script, scriptService);
         }
     }
 }

modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestReindexAction.java

@@ -41,6 +41,7 @@ import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.script.Script;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -74,8 +75,9 @@ public class RestReindexAction extends AbstractBaseReindexRestHandler<ReindexReq
             request.setRemoteInfo(buildRemoteInfo(source));
             XContentBuilder builder = XContentFactory.contentBuilder(parser.contentType());
             builder.map(source);
-            try (XContentParser innerParser = parser.contentType().xContent()
-                    .createParser(parser.getXContentRegistry(), parser.getDeprecationHandler(), builder.bytes().streamInput())) {
+            try (InputStream stream = builder.bytes().streamInput();
+                 XContentParser innerParser = parser.contentType().xContent()
+                     .createParser(parser.getXContentRegistry(), parser.getDeprecationHandler(), stream)) {
                 request.getSearchRequest().source().parseXContent(innerParser);
             }
         };

modules/reindex/src/main/java/org/elasticsearch/index/reindex/TransportReindexAction.java

@@ -72,6 +72,7 @@ import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -338,9 +339,9 @@ public class TransportReindexAction extends HandledTransportAction<ReindexReques
             final XContentType mainRequestXContentType = mainRequest.getDestination().getContentType();
             if (mainRequestXContentType != null && doc.getXContentType() != mainRequestXContentType) {
                 // we need to convert
-                try (XContentParser parser = sourceXContentType.xContent()
-                        .createParser(NamedXContentRegistry.EMPTY,
-                            DeprecationHandler.THROW_UNSUPPORTED_OPERATION, doc.getSource().streamInput());
+                try (InputStream stream = doc.getSource().streamInput();
+                     XContentParser parser = sourceXContentType.xContent()
+                         .createParser(NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION, stream);
                      XContentBuilder builder = XContentBuilder.builder(mainRequestXContentType.xContent())) {
                     parser.nextToken();
                     builder.copyCurrentStructure(parser);

server/src/main/java/org/elasticsearch/action/admin/indices/mapping/put/PutMappingRequest.java

@@ -40,6 +40,7 @@ import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.Index;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.util.Arrays;
 import java.util.Map;
@@ -323,7 +324,9 @@ public class PutMappingRequest extends AcknowledgedRequest<PutMappingRequest> im
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         if (source != null) {
-            builder.rawValue(new BytesArray(source).streamInput(), XContentType.JSON);
+            try (InputStream stream = new BytesArray(source).streamInput()) {
+                builder.rawValue(stream, XContentType.JSON);
+            }
         } else {
             builder.startObject().endObject();
         }

server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java

@@ -49,6 +49,7 @@ import org.elasticsearch.index.VersionType;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -305,9 +306,9 @@ public class BulkRequest extends ActionRequest implements CompositeIndicesReques
 
             // now parse the action
             // EMPTY is safe here because we never call namedObject
-            try (XContentParser parser = xContent
-                    .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE,
-                        data.slice(from, nextMarker - from).streamInput())) {
+            try (InputStream stream = data.slice(from, nextMarker - from).streamInput();
+                 XContentParser parser = xContent
+                     .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, stream)) {
                 // move pointers
                 from = nextMarker + 1;
 
@@ -431,8 +432,9 @@ public class BulkRequest extends ActionRequest implements CompositeIndicesReques
                                 .routing(routing)
                                 .parent(parent);
                         // EMPTY is safe here because we never call namedObject
-                        try (XContentParser sliceParser = xContent.createParser(NamedXContentRegistry.EMPTY,
-                                LoggingDeprecationHandler.INSTANCE, sliceTrimmingCarriageReturn(data, from, nextMarker, xContentType).streamInput())) {
+                        try (InputStream dataStream = sliceTrimmingCarriageReturn(data, from, nextMarker, xContentType).streamInput();
+                             XContentParser sliceParser = xContent.createParser(NamedXContentRegistry.EMPTY,
+                                 LoggingDeprecationHandler.INSTANCE, dataStream)) {
                             updateRequest.fromXContent(sliceParser);
                         }
                         if (fetchSourceContext != null) {

server/src/main/java/org/elasticsearch/action/search/MultiSearchRequest.java

@@ -36,6 +36,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
@@ -207,9 +208,8 @@ public class MultiSearchRequest extends ActionRequest implements CompositeIndice
             IndicesOptions defaultOptions = SearchRequest.DEFAULT_INDICES_OPTIONS;
             // now parse the action
             if (nextMarker - from > 0) {
-                try (XContentParser parser = xContent
-                        .createParser(registry, LoggingDeprecationHandler.INSTANCE,
-                            data.slice(from, nextMarker - from).streamInput())) {
+                try (InputStream stream = data.slice(from, nextMarker - from).streamInput();
+                     XContentParser parser = xContent.createParser(registry, LoggingDeprecationHandler.INSTANCE, stream)) {
                     Map<String, Object> source = parser.map();
                     for (Map.Entry<String, Object> entry : source.entrySet()) {
                         Object value = entry.getValue();
@@ -245,7 +245,8 @@ public class MultiSearchRequest extends ActionRequest implements CompositeIndice
                 break;
             }
             BytesReference bytes = data.slice(from, nextMarker - from);
-            try (XContentParser parser = xContent.createParser(registry, LoggingDeprecationHandler.INSTANCE, bytes.streamInput())) {
+            try (InputStream stream = bytes.streamInput();
+                 XContentParser parser = xContent.createParser(registry, LoggingDeprecationHandler.INSTANCE, stream)) {
                 consumer.accept(searchRequest, parser);
             }
             // move pointers

server/src/main/java/org/elasticsearch/action/termvectors/TermVectorsFields.java

@@ -130,21 +130,21 @@ public final class TermVectorsFields extends Fields {
      * @param termVectors Stores the actual term vectors as a {@link BytesRef}.
      */
     public TermVectorsFields(BytesReference headerRef, BytesReference termVectors) throws IOException {
-        StreamInput header = headerRef.streamInput();
-        fieldMap = new ObjectLongHashMap<>();
-        // here we read the header to fill the field offset map
-        String headerString = header.readString();
-        assert headerString.equals("TV");
-        int version = header.readInt();
-        assert version == -1;
-        hasTermStatistic = header.readBoolean();
-        hasFieldStatistic = header.readBoolean();
-        hasScores = header.readBoolean();
-        final int numFields = header.readVInt();
-        for (int i = 0; i < numFields; i++) {
-            fieldMap.put((header.readString()), header.readVLong());
+        try (StreamInput header = headerRef.streamInput()) {
+            fieldMap = new ObjectLongHashMap<>();
+            // here we read the header to fill the field offset map
+            String headerString = header.readString();
+            assert headerString.equals("TV");
+            int version = header.readInt();
+            assert version == -1;
+            hasTermStatistic = header.readBoolean();
+            hasFieldStatistic = header.readBoolean();
+            hasScores = header.readBoolean();
+            final int numFields = header.readVInt();
+            for (int i = 0; i < numFields; i++) {
+                fieldMap.put((header.readString()), header.readVLong());
+            }
         }
-        header.close();
         // reference to the term vector data
         this.termVectors = termVectors;
     }

server/src/main/java/org/elasticsearch/common/xcontent/XContentBuilder.java

@@ -987,7 +987,9 @@ public final class XContentBuilder implements Releasable, Flushable {
      */
     @Deprecated
     public XContentBuilder rawField(String name, BytesReference value) throws IOException {
-        generator.writeRawField(name, value.streamInput());
+        try (InputStream stream = value.streamInput()) {
+            generator.writeRawField(name, stream);
+        }
         return this;
     }
 
@@ -995,7 +997,9 @@ public final class XContentBuilder implements Releasable, Flushable {
      * Writes a raw field with the given bytes as the value
      */
     public XContentBuilder rawField(String name, BytesReference value, XContentType contentType) throws IOException {
-        generator.writeRawField(name, value.streamInput(), contentType);
+        try (InputStream stream = value.streamInput()) {
+            generator.writeRawField(name, stream, contentType);
+        }
         return this;
     }
 

server/src/main/java/org/elasticsearch/common/xcontent/XContentHelper.java

@@ -106,7 +106,9 @@ public class XContentHelper {
                 input = bytes.streamInput();
             }
             contentType = xContentType != null ? xContentType : XContentFactory.xContentType(input);
-            return new Tuple<>(Objects.requireNonNull(contentType), convertToMap(XContentFactory.xContent(contentType), input, ordered));
+            try (InputStream stream = input) {
+                return new Tuple<>(Objects.requireNonNull(contentType), convertToMap(XContentFactory.xContent(contentType), stream, ordered));
+            }
         } catch (IOException e) {
             throw new ElasticsearchParseException("Failed to parse content to map", e);
         }
@@ -163,8 +165,9 @@ public class XContentHelper {
         }
 
         // It is safe to use EMPTY here because this never uses namedObject
-        try (XContentParser parser = XContentFactory.xContent(xContentType).createParser(NamedXContentRegistry.EMPTY,
-                DeprecationHandler.THROW_UNSUPPORTED_OPERATION, bytes.streamInput())) {
+        try (InputStream stream = bytes.streamInput();
+             XContentParser parser = XContentFactory.xContent(xContentType).createParser(NamedXContentRegistry.EMPTY,
+                 DeprecationHandler.THROW_UNSUPPORTED_OPERATION, stream)) {
             parser.nextToken();
             XContentBuilder builder = XContentFactory.jsonBuilder();
             if (prettyPrint) {
@@ -376,8 +379,9 @@ public class XContentHelper {
     public static void writeRawField(String field, BytesReference source, XContentBuilder builder, ToXContent.Params params) throws IOException {
         Compressor compressor = CompressorFactory.compressor(source);
         if (compressor != null) {
-            InputStream compressedStreamInput = compressor.streamInput(source.streamInput());
-            builder.rawField(field, compressedStreamInput);
+            try (InputStream compressedStreamInput = compressor.streamInput(source.streamInput())) {
+                builder.rawField(field, compressedStreamInput);
+            }
         } else {
             builder.rawField(field, source);
         }
@@ -392,8 +396,9 @@ public class XContentHelper {
         Objects.requireNonNull(xContentType);
         Compressor compressor = CompressorFactory.compressor(source);
         if (compressor != null) {
-            InputStream compressedStreamInput = compressor.streamInput(source.streamInput());
-            builder.rawField(field, compressedStreamInput, xContentType);
+            try (InputStream compressedStreamInput = compressor.streamInput(source.streamInput())) {
+                builder.rawField(field, compressedStreamInput, xContentType);
+            }
         } else {
             builder.rawField(field, source, xContentType);
         }

server/src/main/java/org/elasticsearch/index/query/functionscore/DecayFunctionBuilder.java

@@ -53,6 +53,7 @@ import org.elasticsearch.index.query.QueryShardContext;
 import org.elasticsearch.search.MultiValueMode;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.Objects;
 
 public abstract class DecayFunctionBuilder<DFB extends DecayFunctionBuilder<DFB>> extends ScoreFunctionBuilder<DFB> {
@@ -182,8 +183,9 @@ public abstract class DecayFunctionBuilder<DFB extends DecayFunctionBuilder<DFB>
     protected ScoreFunction doToFunction(QueryShardContext context) throws IOException {
         AbstractDistanceScoreFunction scoreFunction;
         // EMPTY is safe because parseVariable doesn't use namedObject
-        try (XContentParser parser = XContentFactory.xContent(functionBytes)
-                .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, functionBytes.streamInput())) {
+        try (InputStream stream = functionBytes.streamInput();
+             XContentParser parser = XContentFactory.xContent(functionBytes)
+                 .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, stream)) {
             scoreFunction = parseVariable(fieldName, parser, context, multiValueMode);
         }
         return scoreFunction;

server/src/main/java/org/elasticsearch/rest/RestRequest.java

@@ -37,6 +37,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.SocketAddress;
 import java.util.Collections;
 import java.util.HashMap;
@@ -385,8 +386,9 @@ public abstract class RestRequest implements ToXContent.Params {
             Tuple<XContentType, BytesReference> tuple = contentOrSourceParam();
             BytesReference content = tuple.v2();
             XContentType xContentType = tuple.v1();
-            try (XContentParser parser = xContentType.xContent()
-                    .createParser(xContentRegistry, LoggingDeprecationHandler.INSTANCE, content.streamInput())) {
+            try (InputStream stream = content.streamInput();
+                 XContentParser parser = xContentType.xContent()
+                     .createParser(xContentRegistry, LoggingDeprecationHandler.INSTANCE, stream)) {
                 withParser.accept(parser);
             }
         } else {

server/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java

@@ -36,6 +36,7 @@ import org.elasticsearch.rest.action.RestResponseListener;
 import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
 
 import java.io.IOException;
+import java.io.InputStream;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.HEAD;
@@ -84,7 +85,9 @@ public class RestGetSourceAction extends BaseRestHandler {
                             return new BytesRestResponse(NOT_FOUND, builder);
                         } else {
                             final BytesReference source = response.getSourceInternal();
-                            builder.rawValue(source.streamInput(), XContentFactory.xContentType(source));
+                            try (InputStream stream = source.streamInput()) {
+                                builder.rawValue(stream, XContentFactory.xContentType(source));
+                            }
                             return new BytesRestResponse(OK, builder);
                         }
                     }

server/src/main/java/org/elasticsearch/script/Script.java

@@ -40,6 +40,7 @@ import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.util.Collections;
 import java.util.HashMap;
@@ -282,8 +283,11 @@ public final class Script implements ToXContentObject, Writeable {
             builder.startObject();
             settings.toXContent(builder, ToXContent.EMPTY_PARAMS);
             builder.endObject();
-            return parse(JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
-                LoggingDeprecationHandler.INSTANCE, builder.bytes().streamInput()));
+            try (InputStream stream = builder.bytes().streamInput();
+                 XContentParser parser = JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
+                     LoggingDeprecationHandler.INSTANCE, stream)) {
+                return parse(parser);
+            }
         } catch (IOException e) {
             // it should not happen since we are not actually reading from a stream but an in-memory byte[]
             throw new IllegalStateException(e);

server/src/main/java/org/elasticsearch/script/StoredScriptSource.java

@@ -43,6 +43,7 @@ import org.elasticsearch.common.xcontent.XContentParser.Token;
 import org.elasticsearch.common.xcontent.XContentType;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.util.Collections;
 import java.util.HashMap;
@@ -243,8 +244,9 @@ public class StoredScriptSource extends AbstractDiffable<StoredScriptSource> imp
      * @return        The parsed {@link StoredScriptSource}.
      */
     public static StoredScriptSource parse(BytesReference content, XContentType xContentType) {
-        try (XContentParser parser = xContentType.xContent()
-                .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, content.streamInput())) {
+        try (InputStream stream = content.streamInput();
+             XContentParser parser = xContentType.xContent()
+                 .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, stream)) {
             Token token = parser.nextToken();
 
             if (token != Token.START_OBJECT) {

server/src/main/java/org/elasticsearch/tasks/RawTaskStatus.java

@@ -27,6 +27,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.Map;
 
 import static java.util.Objects.requireNonNull;
@@ -58,7 +59,9 @@ public class RawTaskStatus implements Task.Status {
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        return builder.rawValue(status.streamInput(), XContentFactory.xContentType(status));
+        try (InputStream stream = status.streamInput()) {
+            return builder.rawValue(stream, XContentFactory.xContentType(status));
+        }
     }
 
     @Override