From 3ad04403f2cdd2b8a5b5e5b48f2f0f763d3f90d2 Mon Sep 17 00:00:00 2001
From: Laura Trotta <laura.trotta@elastic.co>
Date: Fri, 19 Dec 2025 15:39:31 +0100
Subject: [PATCH] avoid double producer

---
 .../transport/rest5_client/MultiBufferEntity.java | 15 +++++++++++++--
 .../rest5_client/low_level/Rest5Client.java       | 11 +++++++++--
 2 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/java-client/src/main/java/co/elastic/clients/transport/rest5_client/MultiBufferEntity.java b/java-client/src/main/java/co/elastic/clients/transport/rest5_client/MultiBufferEntity.java
index 674e85da83..742753036b 100644
--- a/java-client/src/main/java/co/elastic/clients/transport/rest5_client/MultiBufferEntity.java
+++ b/java-client/src/main/java/co/elastic/clients/transport/rest5_client/MultiBufferEntity.java
@@ -22,7 +22,7 @@
 import co.elastic.clients.util.NoCopyByteArrayOutputStream;
 import org.apache.hc.core5.http.ContentType;
 import org.apache.hc.core5.http.io.entity.AbstractHttpEntity;
-import org.apache.hc.core5.http.nio.AsyncDataProducer;
+import org.apache.hc.core5.http.nio.AsyncEntityProducer;
 import org.apache.hc.core5.http.nio.DataStreamChannel;
 
 import java.io.IOException;
@@ -32,20 +32,24 @@
 import java.nio.channels.Channels;
 import java.nio.channels.WritableByteChannel;
 import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * An HTTP entity based on a sequence of byte buffers.
  */
-class MultiBufferEntity extends AbstractHttpEntity implements AsyncDataProducer {
+class MultiBufferEntity extends AbstractHttpEntity implements AsyncEntityProducer {
 
     private final Iterable<ByteBuffer> buffers;
 
     private Iterator<ByteBuffer> iterator;
     private volatile ByteBuffer currentBuffer;
 
+    private final AtomicReference<Exception> exceptionRef;
+
     MultiBufferEntity(Iterable<ByteBuffer> buffers, ContentType contentType) {
         super(contentType,null,true);
         this.buffers = buffers;
+        this.exceptionRef = new AtomicReference<>();
         init();
     }
 
@@ -69,6 +73,13 @@ public boolean isRepeatable() {
         return true;
     }
 
+    @Override
+    public void failed(Exception cause) {
+        if (exceptionRef.compareAndSet(null, cause)) {
+            releaseResources();
+        }
+    }
+
     @Override
     public long getContentLength() {
         // Use chunked encoding
diff --git a/rest5-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java b/rest5-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java
index 859b66f549..66bc94aa36 100644
--- a/rest5-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java
+++ b/rest5-client/src/main/java/co/elastic/clients/transport/rest5_client/low_level/Rest5Client.java
@@ -45,6 +45,7 @@
 import org.apache.hc.core5.http.HttpRequest;
 import org.apache.hc.core5.http.ProtocolException;
 import org.apache.hc.core5.http.message.RequestLine;
+import org.apache.hc.core5.http.nio.AsyncEntityProducer;
 import org.apache.hc.core5.http.nio.AsyncRequestProducer;
 import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
 import org.apache.hc.core5.http.nio.support.AsyncRequestBuilder;
@@ -409,7 +410,7 @@ private void performRequestAsync(
             final RequestContext context;
             context = request.createContextForNextAttempt(nodes.next());
             Future<ClassicHttpResponse> futureRef = client.execute(context.requestProducer, context.asyncResponseConsumer, context.context,
-                new FutureCallback<ClassicHttpResponse>() {
+                new FutureCallback<>() {
                     @Override
                     public void completed(ClassicHttpResponse httpResponse) {
                         try {
@@ -877,7 +878,13 @@ private static class RequestContext {
                 .setHeaders(request.httpRequest.getHeaders());
 
             if (request.httpRequest.getEntity() != null) {
-                builder.setEntity(new BasicAsyncEntityProducer(request.httpRequest.getEntity()));
+                HttpEntity entity = request.httpRequest.getEntity();
+                if (entity instanceof AsyncEntityProducer) {
+                    builder.setEntity((AsyncEntityProducer) request.httpRequest.getEntity());
+                }
+                else {
+                    builder.setEntity(new BasicAsyncEntityProducer(entity));
+                }
             }
 
             this.requestProducer = builder.build();