Cat APIs should not close the stream obtained from the channel

The cat APIs and rest tables would obtain a stream from the RestChannel, which happened to be a
ReleasableBytesStreamOutput. These APIs used the stream to write content to, closed the stream,
and then tried to send a response. After #23941 was merged, closing the stream meant that the bytes
were released for use elsewhere. This caused occasional corruption of the response when the bytes
were used prior to the response being sent.

This commit changes these two usages to wrap the stream obtained from the channel in a flush on
close stream so that the bytes are still reserved until the message is sent.

Author: jaymode

core/src/main/java/org/elasticsearch/rest/action/cat/AbstractCatAction.java

@@ -20,8 +20,9 @@
 
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Table;
+import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.io.UTF8StreamWriter;
-import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.io.stream.BytesStream;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.BytesRestResponse;
@@ -56,7 +57,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             return channel -> {
                 Table table = getTableWithHeader(request);
                 int[] width = buildHelpWidths(table, request);
-                BytesStreamOutput bytesOutput = channel.bytesOutput();
+                BytesStream bytesOutput = Streams.flushOnCloseStream(channel.bytesOutput());
                 UTF8StreamWriter out = new UTF8StreamWriter().setOutput(bytesOutput);
                 for (Table.Cell cell : table.getHeaders()) {
                     // need to do left-align always, so create new cells

core/src/main/java/org/elasticsearch/rest/action/cat/RestTable.java

@@ -22,8 +22,9 @@
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.Table;
+import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.io.UTF8StreamWriter;
-import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.io.stream.BytesStream;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.SizeValue;
@@ -82,7 +83,7 @@ public static RestResponse buildTextPlainResponse(Table table, RestChannel chann
         List<DisplayHeader> headers = buildDisplayHeaders(table, request);
         int[] width = buildWidths(table, request, verbose, headers);
 
-        BytesStreamOutput bytesOut = channel.bytesOutput();
+        BytesStream bytesOut = Streams.flushOnCloseStream(channel.bytesOutput());
         UTF8StreamWriter out = new UTF8StreamWriter().setOutput(bytesOut);
         int lastHeader = headers.size() - 1;
         if (verbose) {