Closing a ReleasableBytesStreamOutput closes the underlying BigArray (#23941)

This commit makes closing a ReleasableBytesStreamOutput release the underlying BigArray so
that we can use try-with-resources with these streams and avoid leaking memory by not returning
the BigArray. As part of this change, the ReleasableBytesStreamOutput adds protection to only
release the BigArray once.

In order to make some of the changes cleaner, the ReleasableBytesStream interface has been
removed. The BytesStream interface is changed to a abstract class so that we can use it as a
useable return type for a new method, Streams#flushOnCloseStream. This new method wraps a
given stream and overrides the close method so that the stream is simply flushed and not closed.
This behavior is used in the TcpTransport when compression is used with a
ReleasableBytesStreamOutput as we need to close the compressed stream to ensure all of the data
is written from this stream. Closing the compressed stream will try to close the underlying stream
but we only want to flush so that all of the written bytes are available.

Additionally, an error message method added in the BytesRestResponse did not use a builder
provided by the channel and instead created its own JSON builder. This changes that method to use
the channel builder and in turn the bytes stream output that is managed by the channel.

Note, this commit differs from 6bfecdf921a1941b48273d76551872df4062cfae in that it updates
ReleasableBytesStreamOutput to handle the case of the BigArray decreasing in size, which changes
the reference to the BigArray. When the reference is changed, the releasable needs to be updated
otherwise there could be a leak of bytes and corruption of data in unrelated streams.

This reverts commit afd45c14327cd0f8d155e5ac9740f48e8e39b09c, which reverted #23572.

1 files changed, 51 insertions, 0 deletions

diff --git a/core/src/test/java/org/elasticsearch/common/io/stream/ReleasableBytesStreamOutputTests.java b/core/src/test/java/org/elasticsearch/common/io/stream/ReleasableBytesStreamOutputTests.java
new file mode 100644
index 0000000000..557721a024
--- /dev/null
+++ b/core/src/test/java/org/elasticsearch/common/io/stream/ReleasableBytesStreamOutputTests.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.io.stream;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
+import org.elasticsearch.test.ESTestCase;
+
+import java.io.IOException;
+
+public class ReleasableBytesStreamOutputTests extends ESTestCase {
+
+    public void testRelease() throws Exception {
+        MockBigArrays mockBigArrays =
+            new MockBigArrays(Settings.EMPTY, new NoneCircuitBreakerService());
+        try (ReleasableBytesStreamOutput output =
+                 getRandomReleasableBytesStreamOutput(mockBigArrays)) {
+            output.writeBoolean(randomBoolean());
+        }
+        MockBigArrays.ensureAllArraysAreReleased();
+    }
+
+    private ReleasableBytesStreamOutput getRandomReleasableBytesStreamOutput(
+                                                MockBigArrays mockBigArrays) throws IOException {
+        ReleasableBytesStreamOutput output = new ReleasableBytesStreamOutput(mockBigArrays);
+        if (randomBoolean()) {
+            for (int i = 0; i < scaledRandomIntBetween(1, 32); i++) {
+                output.write(randomByte());
+            }
+        }
+        return output;
+    }
+}