Make buffer limit configurable in HeapBufferedConsumerFactory (#23970)

The buffer limit should have been configurable already, but the factory constructor is package private so it is truly configurable only from the org.elasticsearch.client package. Also the HttpAsyncResponseConsumerFactory interface was package private, so it could only be implemented from the org.elasticsearch.client package.

Closes #23958

buildSrc/src/main/resources/checkstyle_suppressions.xml

@@ -42,6 +42,7 @@
   <suppress files="client[/\\]test[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]client[/\\]RestClientTestUtil.java" checks="LineLength" />
   <suppress files="client[/\\]rest[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]client[/\\]RestClientTests.java" checks="LineLength" />
   <suppress files="client[/\\]rest[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]client[/\\]SyncResponseListenerTests.java" checks="LineLength" />
+  <suppress files="client[/\\]rest[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]client[/\\]HeapBufferedAsyncResponseConsumerTests.java" checks="LineLength" />
   <suppress files="client[/\\]rest-high-level[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]client[/\\]Request.java" checks="LineLength" />
   <suppress files="client[/\\]rest-high-level[/\\]src[/\\]main[/\\]java[/\\]org[/\\]elasticsearch[/\\]client[/\\]RestHighLevelClient.java" checks="LineLength" />
   <suppress files="client[/\\]rest-high-level[/\\]src[/\\]test[/\\]java[/\\]org[/\\]elasticsearch[/\\]client[/\\]CrudIT.java" checks="LineLength" />

client/rest/src/main/java/org/elasticsearch/client/HttpAsyncResponseConsumerFactory.java

@@ -29,7 +29,7 @@ import static org.elasticsearch.client.HttpAsyncResponseConsumerFactory.HeapBuff
  * consumer object. Users can implement this interface and pass their own instance to the specialized
  * performRequest methods that accept an {@link HttpAsyncResponseConsumerFactory} instance as argument.
  */
-interface HttpAsyncResponseConsumerFactory {
+public interface HttpAsyncResponseConsumerFactory {
 
     /**
      * Creates the default type of {@link HttpAsyncResponseConsumer}, based on heap buffering with a buffer limit of 100MB.
@@ -53,7 +53,7 @@ interface HttpAsyncResponseConsumerFactory {
 
         private final int bufferLimit;
 
-        HeapBufferedResponseConsumerFactory(int bufferLimitBytes) {
+        public HeapBufferedResponseConsumerFactory(int bufferLimitBytes) {
             this.bufferLimit = bufferLimitBytes;
         }
 

client/rest/src/test/java/org/elasticsearch/client/HeapBufferedAsyncResponseConsumerTests.java

@@ -24,19 +24,24 @@ import org.apache.http.HttpEntity;
 import org.apache.http.HttpResponse;
 import org.apache.http.ProtocolVersion;
 import org.apache.http.StatusLine;
-import org.apache.http.entity.BasicHttpEntity;
 import org.apache.http.entity.ContentType;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.message.BasicHttpResponse;
 import org.apache.http.message.BasicStatusLine;
 import org.apache.http.nio.ContentDecoder;
 import org.apache.http.nio.IOControl;
+import org.apache.http.nio.protocol.HttpAsyncResponseConsumer;
 import org.apache.http.protocol.HttpContext;
 
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Modifier;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -97,6 +102,26 @@ public class HeapBufferedAsyncResponseConsumerTests extends RestClientTestCase {
         bufferLimitTest(consumer, bufferLimit);
     }
 
+    public void testCanConfigureHeapBufferLimitFromOutsidePackage() throws ClassNotFoundException, NoSuchMethodException,
+            IllegalAccessException, InvocationTargetException, InstantiationException {
+        int bufferLimit = randomIntBetween(1, Integer.MAX_VALUE);
+        //we use reflection to make sure that the class can be instantiated from the outside, and the constructor is public
+        Constructor<?> constructor = HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory.class.getConstructor(Integer.TYPE);
+        assertEquals(Modifier.PUBLIC, constructor.getModifiers() & Modifier.PUBLIC);
+        Object object = constructor.newInstance(bufferLimit);
+        assertThat(object, instanceOf(HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory.class));
+        HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory consumerFactory =
+                (HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory) object;
+        HttpAsyncResponseConsumer<HttpResponse> consumer = consumerFactory.createHttpAsyncResponseConsumer();
+        assertThat(consumer, instanceOf(HeapBufferedAsyncResponseConsumer.class));
+        HeapBufferedAsyncResponseConsumer bufferedAsyncResponseConsumer = (HeapBufferedAsyncResponseConsumer) consumer;
+        assertEquals(bufferLimit, bufferedAsyncResponseConsumer.getBufferLimit());
+    }
+
+    public void testHttpAsyncResponseConsumerFactoryVisibility() throws ClassNotFoundException {
+        assertEquals(Modifier.PUBLIC, HttpAsyncResponseConsumerFactory.class.getModifiers() & Modifier.PUBLIC);
+    }
+
     private static void bufferLimitTest(HeapBufferedAsyncResponseConsumer consumer, int bufferLimit) throws Exception {
         ProtocolVersion protocolVersion = new ProtocolVersion("HTTP", 1, 1);
         StatusLine statusLine = new BasicStatusLine(protocolVersion, 200, "OK");