From 3c26e656cb140f3a3ef3255b6345a5c02a9ca163 Mon Sep 17 00:00:00 2001 From: Ryan Dew Date: Tue, 22 Sep 2026 19:01:34 -0400 Subject: [PATCH 1/3] MLE-32970 Fix multipart streaming input to data services --- .../marklogic/client/impl/OkHttpServices.java | 11 ++- .../client/impl/StreamingOutputImpl.java | 11 ++- .../okhttp/RetryIOExceptionInterceptor.java | 7 +- .../client/impl/StreamingOutputImplTest.java | 84 +++++++++++++++++++ 4 files changed, 109 insertions(+), 4 deletions(-) create mode 100644 marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java b/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java index 1de3b0bffb..e4e8d87e04 100644 --- a/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java +++ b/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java @@ -5194,7 +5194,7 @@ static private List getPartList(MimeMultipart multipart) { } } - static private class ObjectRequestBody extends RequestBody implements RetryableRequestBody { + static class ObjectRequestBody extends RequestBody implements RetryableRequestBody { private Object obj; private MediaType contentType; @@ -5236,6 +5236,15 @@ public boolean isRetryable() { // InputStreams cannot be retried as they are consumed on first read. return !(obj instanceof InputStream); } + + @Override + public boolean isOneShot() { + // Declares an InputStream-backed body as one-shot via OkHttp's own contract so that when this body is + // nested inside another RequestBody (e.g. a MultipartBody part), OkHttp and RetryIOExceptionInterceptor + // both recognize the outer body as non-retryable too; isRetryable() above is only checked when this is the + // top-level request body and would otherwise be bypassed for a nested InputStream part. + return obj instanceof InputStream; + } } // API First Changes diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java b/marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java index 5fd30e6d05..11e909ccfc 100644 --- a/marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java +++ b/marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved. + * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved. */ package com.marklogic.client.impl; @@ -53,4 +53,13 @@ public boolean isRetryable() { // Added in 8.0.0; streaming output cannot be retried as the stream is consumed on first write. return false; } + + @Override + public boolean isOneShot() { + // Declares this body as one-shot via OkHttp's own contract so that when this body is nested inside another + // RequestBody (e.g. a MultipartBody part for a Data Services call), OkHttp and RetryIOExceptionInterceptor + // both recognize the outer body as non-retryable too; isRetryable() above is only checked when this is the + // top-level request body and would otherwise be bypassed for a nested streaming part. + return true; + } } diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/impl/okhttp/RetryIOExceptionInterceptor.java b/marklogic-client-api/src/main/java/com/marklogic/client/impl/okhttp/RetryIOExceptionInterceptor.java index b2f57e0c3e..d2517828c7 100644 --- a/marklogic-client-api/src/main/java/com/marklogic/client/impl/okhttp/RetryIOExceptionInterceptor.java +++ b/marklogic-client-api/src/main/java/com/marklogic/client/impl/okhttp/RetryIOExceptionInterceptor.java @@ -1,11 +1,12 @@ /* - * Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved. + * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved. */ package com.marklogic.client.impl.okhttp; import com.marklogic.client.MarkLogicIOException; import okhttp3.Interceptor; import okhttp3.Request; +import okhttp3.RequestBody; import okhttp3.Response; import org.slf4j.Logger; @@ -40,8 +41,10 @@ public RetryIOExceptionInterceptor(int maxRetries, long initialDelayMs, double b @Override public Response intercept(Chain chain) throws IOException { Request request = chain.request(); + RequestBody requestBody = request.body(); - if (request.body() instanceof RetryableRequestBody body && !body.isRetryable()) { + if (requestBody != null && (requestBody.isOneShot() || + requestBody instanceof RetryableRequestBody body && !body.isRetryable())) { return chain.proceed(request); } diff --git a/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java b/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java new file mode 100644 index 0000000000..55de07d49d --- /dev/null +++ b/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java @@ -0,0 +1,84 @@ +/* + * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved. + */ +package com.marklogic.client.impl; + +import com.marklogic.client.impl.okhttp.RetryIOExceptionInterceptor; +import okhttp3.MediaType; +import okhttp3.MultipartBody; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class StreamingOutputImplTest { + + @Test + void multipartBodyWithStreamingPartIsNotRetried() { + StreamingOutputImpl streamingBody = new StreamingOutputImpl( + outputStream -> outputStream.write("content".getBytes(StandardCharsets.UTF_8)), + null, + MediaType.get("application/xml") + ); + MultipartBody multipartBody = new MultipartBody.Builder() + .setType(MultipartBody.FORM) + .addFormDataPart("body", null, streamingBody) + .build(); + assertTrue(multipartBody.isOneShot()); + + AtomicInteger attempts = new AtomicInteger(); + OkHttpClient client = new OkHttpClient.Builder() + .addInterceptor(new RetryIOExceptionInterceptor(3, 0, 1, 0)) + .addInterceptor(chain -> { + attempts.incrementAndGet(); + throw new IOException("unexpected end of stream"); + }) + .build(); + Request request = new Request.Builder() + .url("http://localhost/") + .post(multipartBody) + .build(); + + assertThrows(IOException.class, () -> client.newCall(request).execute()); + assertEquals(1, attempts.get()); + } + + @Test + void multipartBodyWithInputStreamPartIsNotRetried() { + InputStream inputStream = new ByteArrayInputStream("content".getBytes(StandardCharsets.UTF_8)); + OkHttpServices.ObjectRequestBody inputStreamBody = + new OkHttpServices.ObjectRequestBody(inputStream, MediaType.get("application/xml")); + MultipartBody multipartBody = new MultipartBody.Builder() + .setType(MultipartBody.FORM) + .addFormDataPart("body", null, inputStreamBody) + .build(); + assertTrue(multipartBody.isOneShot(), + "A multipart body containing an InputStream-backed part must be one-shot so the retry interceptor does " + + "not replay an already-consumed stream."); + + AtomicInteger attempts = new AtomicInteger(); + OkHttpClient client = new OkHttpClient.Builder() + .addInterceptor(new RetryIOExceptionInterceptor(3, 0, 1, 0)) + .addInterceptor(chain -> { + attempts.incrementAndGet(); + throw new IOException("unexpected end of stream"); + }) + .build(); + Request request = new Request.Builder() + .url("http://localhost/") + .post(multipartBody) + .build(); + + assertThrows(IOException.class, () -> client.newCall(request).execute()); + assertEquals(1, attempts.get()); + } +} From cdb2f2c96d6d63bf892e13b9dfb0dffbc54db808 Mon Sep 17 00:00:00 2001 From: Ryan Dew Date: Thu, 8 Oct 2026 11:06:32 -0700 Subject: [PATCH 2/3] MLE-32953 Make StreamingOutputImpl retry decision reflect handle resendability StreamingOutputImpl previously hardcoded isRetryable()=false / isOneShot()=true for every wrapped OutputStreamSender, even when the originating write handle was fully buffered and resendable (e.g. JacksonHandle, StringHandle). This meant any request body built from such a handle -- including Row Manager plan execution, document writes, and Data Services parameters -- was needlessly blocked from being retried by RetryIOExceptionInterceptor after a transient connection error, even though the content could safely be written again. Thread each call site's already-known isResendable()/handleBase.isResendable() value into StreamingOutputImpl instead of assuming non-resendable: - writeDocumentImpl-style document writes - putPostValueImpl - postResource / putResource (used by RowManagerImpl for Optic plan execution) - postIteratedResourceImpl (multipart-mixed eval/invoke style resources) - addParts multipart builder - makeRequestBodyForContent and BaseProxy's makeRequestBody (Data Services single-node parameters) Call sites without an existing resendable concept (structured query search, alert match) keep the prior conservative false/one-shot default via new 3-arg doPost/doPut overloads, so behavior is unchanged where resendability was never determined. Added StreamingOutputImplTest cases covering: a resendable body being retried after a simulated connection failure, and a non-resendable part inside a MultipartBody still propagating isOneShot()=true so it is never retried. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../marklogic/client/impl/OkHttpServices.java | 31 +++++++--- .../client/impl/StreamingOutputImpl.java | 24 ++++++- .../client/impl/StreamingOutputImplTest.java | 62 +++++++++++++++++++ 3 files changed, 104 insertions(+), 13 deletions(-) diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java b/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java index e4e8d87e04..257d8c81dd 100644 --- a/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java +++ b/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java @@ -1229,7 +1229,7 @@ private TemporalDescriptor putPostDocumentImpl(RequestLogger reqlog, String meth MediaType mediaType = makeType(requestBldr.build().header(HEADER_CONTENT_TYPE)); if (value instanceof OutputStreamSender) { StreamingOutputImpl sentStream = - new StreamingOutputImpl((OutputStreamSender) value, reqlog, mediaType); + new StreamingOutputImpl((OutputStreamSender) value, reqlog, mediaType, isResendable); requestBldr = ("put".equals(method)) ? requestBldr.put(sentStream) : @@ -2457,7 +2457,7 @@ private void putPostValueImpl(RequestLogger reqlog, String method, RequestBody sentValue; if (nextValue instanceof OutputStreamSender) { sentValue = new StreamingOutputImpl( - (OutputStreamSender) nextValue, reqlog, mediaType); + (OutputStreamSender) nextValue, reqlog, mediaType, handle != null && handle.isResendable()); } else { if (reqlog != null && retryContext.getRetry() == 0) { sentValue = new ObjectRequestBody(reqlog.copyContent(nextValue), mediaType); @@ -2831,7 +2831,7 @@ public R putResource(RequestLogger reqlog, Function doPutFunction = new Function() { public Response apply(Request.Builder funcBuilder) { - return doPut(reqlog, funcBuilder, inputBase.sendContent()); + return doPut(reqlog, funcBuilder, inputBase.sendContent(), isResendable); } }; Response response = sendRequestWithRetry(requestBldr, (transaction == null), doPutFunction, resendableConsumer); @@ -2980,7 +2980,7 @@ public void accept(Boolean resendable) { final Object value = inputBase == null ? null : inputBase.sendContent(); Function doPostFunction = new Function() { public Response apply(Request.Builder funcBuilder) { - return doPost(reqlog, funcBuilder, value); + return doPost(reqlog, funcBuilder, value, isResendable); } }; @@ -3573,7 +3573,8 @@ private U postIteratedResourceImpl( Function doPostFunction = requestBuilder -> doPost( reqlog, requestBuilder.header(HEADER_ACCEPT, multipartMixedWithBoundary()), - inputBase.sendContent() + inputBase.sendContent(), + isResendable ); Response response = sendRequestWithRetry(requestBldr, (transaction == null), doPostFunction, resendableConsumer); @@ -3746,13 +3747,18 @@ private Request.Builder makePutWebResource(String path, } private Response doPut(RequestLogger reqlog, Request.Builder requestBldr, Object value) { + // Default to non-resendable for callers that have not determined whether the content can be sent again. + return doPut(reqlog, requestBldr, value, false); + } + + private Response doPut(RequestLogger reqlog, Request.Builder requestBldr, Object value, boolean isResendable) { if (value == null) throw new IllegalArgumentException("Resource write with null value"); if (isFirstRequest() && isStreaming(value)) pingServerBeforeStreaming(0); MediaType mediaType = makeType(requestBldr.build().header(HEADER_CONTENT_TYPE)); if (value instanceof OutputStreamSender) { - requestBldr = requestBldr.put(new StreamingOutputImpl((OutputStreamSender) value, reqlog, mediaType)); + requestBldr = requestBldr.put(new StreamingOutputImpl((OutputStreamSender) value, reqlog, mediaType, isResendable)); } else { if (reqlog != null) { requestBldr = requestBldr.put(new ObjectRequestBody(reqlog.copyContent(value), mediaType)); @@ -3788,6 +3794,11 @@ private Request.Builder makePostWebResource(String path, RequestParameters param } private Response doPost(RequestLogger reqlog, Request.Builder requestBldr, Object value) { + // Default to non-resendable for callers that have not determined whether the content can be sent again. + return doPost(reqlog, requestBldr, value, false); + } + + private Response doPost(RequestLogger reqlog, Request.Builder requestBldr, Object value, boolean isResendable) { if (isFirstRequest() && isStreaming(value)) { pingServerBeforeStreaming(0); } @@ -3799,7 +3810,7 @@ private Response doPost(RequestLogger reqlog, Request.Builder requestBldr, Objec requestBldr = requestBldr.post((MultipartBody) value); } else if (value instanceof OutputStreamSender) { requestBldr = requestBldr - .post(new StreamingOutputImpl((OutputStreamSender) value, reqlog, mediaType)); + .post(new StreamingOutputImpl((OutputStreamSender) value, reqlog, mediaType, isResendable)); } else { if (reqlog != null) { requestBldr = requestBldr.post(new ObjectRequestBody(reqlog.copyContent(value), mediaType)); @@ -4003,7 +4014,7 @@ private boolean addParts( Part bodyPart = null; if (value instanceof OutputStreamSender) { bodyPart = Part.create(partHeaders.build(), new StreamingOutputImpl( - (OutputStreamSender) value, reqlog, mediaType)); + (OutputStreamSender) value, reqlog, mediaType, handleBase.isResendable())); } else { if (reqlog != null) { bodyPart = Part.create(partHeaders.build(), new ObjectRequestBody(reqlog.copyContent(value), mediaType)); @@ -5291,7 +5302,7 @@ public static RequestBody makeRequestBodyForContent(AbstractWriteHandle content) (mimetype != null) ? mimetype : "application/x-unknown-content-type" ); return (content instanceof OutputStreamSender) ? - new StreamingOutputImpl((OutputStreamSender) content, null, mediaType) : + new StreamingOutputImpl((OutputStreamSender) content, null, mediaType, handleBase.isResendable()) : new ObjectRequestBody(HandleAccessor.sendContent(content), mediaType); } @@ -5513,7 +5524,7 @@ private RequestBody makeRequestBody(AbstractWriteHandle document) { (mimetype != null) ? mimetype : "application/x-unknown-content-type" ); return (document instanceof OutputStreamSender) ? - new StreamingOutputImpl((OutputStreamSender) document, null, mediaType) : + new StreamingOutputImpl((OutputStreamSender) document, null, mediaType, handleBase.isResendable()) : new ObjectRequestBody(HandleAccessor.sendContent(document), mediaType); } diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java b/marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java index 11e909ccfc..62cdf697dd 100644 --- a/marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java +++ b/marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java @@ -18,12 +18,27 @@ class StreamingOutputImpl extends RequestBody implements RetryableRequestBody { private OutputStreamSender handle; private RequestLogger logger; private MediaType contentType; + private boolean resendable; + /** + * Assumes the wrapped handle cannot be resent, which is the safe default for callers that have not determined + * whether the underlying content can be sent again (e.g. because it is backed by a one-shot stream). + */ StreamingOutputImpl(OutputStreamSender handle, RequestLogger logger, MediaType contentType) { + this(handle, logger, contentType, false); + } + + /** + * @param resendable whether the wrapped handle's content can be written again if the request must be retried; + * should reflect the originating handle's own {@code isResendable()} value rather than + * always assuming the content is one-shot. + */ + StreamingOutputImpl(OutputStreamSender handle, RequestLogger logger, MediaType contentType, boolean resendable) { super(); this.handle = handle; this.logger = logger; this.contentType = contentType; + this.resendable = resendable; } @Override @@ -50,8 +65,11 @@ public void writeTo(BufferedSink sink) throws IOException { @Override public boolean isRetryable() { - // Added in 8.0.0; streaming output cannot be retried as the stream is consumed on first write. - return false; + // Added in 8.0.0. Originally always returned false, on the assumption that the wrapped stream is consumed + // on first write. That is not true for handles whose content is fully buffered (e.g. JacksonHandle, + // StringHandle) and which report isResendable() == true; for those, the content can safely be written + // again, so we defer to the resendable flag supplied by the caller instead of hardcoding false. + return resendable; } @Override @@ -60,6 +78,6 @@ public boolean isOneShot() { // RequestBody (e.g. a MultipartBody part for a Data Services call), OkHttp and RetryIOExceptionInterceptor // both recognize the outer body as non-retryable too; isRetryable() above is only checked when this is the // top-level request body and would otherwise be bypassed for a nested streaming part. - return true; + return !resendable; } } diff --git a/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java b/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java index 55de07d49d..f92155899c 100644 --- a/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java +++ b/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java @@ -17,6 +17,7 @@ import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -81,4 +82,65 @@ void multipartBodyWithInputStreamPartIsNotRetried() { assertThrows(IOException.class, () -> client.newCall(request).execute()); assertEquals(1, attempts.get()); } + + @Test + void resendableStreamingBodyIsRetried() { + StreamingOutputImpl resendableBody = new StreamingOutputImpl( + outputStream -> outputStream.write("content".getBytes(StandardCharsets.UTF_8)), + null, + MediaType.get("application/xml"), + true + ); + assertTrue(resendableBody.isRetryable(), + "A handle that reports isResendable() == true should allow its StreamingOutputImpl wrapper to be retried."); + assertFalse(resendableBody.isOneShot(), + "A resendable StreamingOutputImpl should not be marked one-shot, so OkHttp permits rewriting its body."); + + AtomicInteger attempts = new AtomicInteger(); + OkHttpClient client = new OkHttpClient.Builder() + .addInterceptor(new RetryIOExceptionInterceptor(3, 0, 1, 0)) + .addInterceptor(chain -> { + if (attempts.incrementAndGet() < 2) { + throw new IOException("unexpected end of stream"); + } + return new okhttp3.Response.Builder() + .request(chain.request()) + .protocol(okhttp3.Protocol.HTTP_1_1) + .code(200) + .message("OK") + .body(okhttp3.ResponseBody.create("", null)) + .build(); + }) + .build(); + Request request = new Request.Builder() + .url("http://localhost/") + .post(resendableBody) + .build(); + + try (okhttp3.Response response = client.newCall(request).execute()) { + assertEquals(200, response.code()); + } catch (IOException e) { + throw new RuntimeException(e); + } + assertEquals(2, attempts.get(), + "The request should have been retried once after the first simulated connection failure."); + } + + @Test + void nonResendableHandleInMultipartIsNotRetried() { + // Mirrors how OkHttpServices wraps a non-resendable handle (e.g. ReaderHandle) in a Data Services multipart + // request: the per-part resendable flag should propagate up through MultipartBody.isOneShot(). + StreamingOutputImpl nonResendablePart = new StreamingOutputImpl( + outputStream -> outputStream.write("content".getBytes(StandardCharsets.UTF_8)), + null, + MediaType.get("application/xml"), + false + ); + MultipartBody multipartBody = new MultipartBody.Builder() + .setType(MultipartBody.FORM) + .addFormDataPart("body", null, nonResendablePart) + .build(); + assertTrue(multipartBody.isOneShot(), + "A multipart request containing even one non-resendable part must remain one-shot overall."); + } } From 4779f8fd33b92ab456d82fe7c942bcaf18f19f30 Mon Sep 17 00:00:00 2001 From: Hemanth Madala Date: Thu, 5 Nov 2026 09:29:35 +0530 Subject: [PATCH 3/3] MLE-32970 Ping before every streaming Data Services request The Java client always sends Accept-Encoding: gzip, and MarkLogic can close a connection after replying to such a request even though OkHttp keeps it pooled. A Data Services call that streams a one-shot body (e.g. a Reader), made right after another call such as RowManager.resultRows(), could be written to that closed connection. The body was consumed before the failure was noticed, so the retry failed with 'Stream closed'. The ping that guards against this only ran for the first request on a thread, and only for digest authentication, so it had already been used up by the earlier call. It now runs before every request that has a streaming part, so a closed connection is found by a ping that has no body to lose. This relies on StreamingOutputImpl reporting one-shot bodies as such: with the ping but without that, a failure after the ping is resent by OkHttp and reports 'Stream closed' again. StreamingPartRequestTest (mock server) checks that the ping happens before each streaming request and that a failed streaming request is not resent. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../marklogic/client/impl/OkHttpServices.java | 9 +- .../client/impl/StreamingPartRequestTest.java | 94 +++++++++++++++++++ 2 files changed, 101 insertions(+), 2 deletions(-) create mode 100644 marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingPartRequestTest.java diff --git a/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java b/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java index 257d8c81dd..4f6db8242a 100644 --- a/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java +++ b/marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java @@ -5409,8 +5409,13 @@ private void executeRequest(CallResponseImpl responseImpl) { }; Function sendRequestFunction = requestBldr -> { - if (isFirstRequest() && hasStreamingPart) { - // Ping the server before streaming; if unavailable, wait and retry the ping + if (hasStreamingPart) { + // Send a body-less ping first when this request streams a one-shot body (e.g. a Reader). + // MarkLogic can close a connection after replying to a request that accepts gzip, which this client + // always sends, yet OkHttp may still reuse it from the pool. A one-shot body written to such a + // connection is lost and cannot be resent, so the ping finds the dead connection first. This must + // happen for every such request, not only the first. If the server is unavailable, wait and retry + // the ping. int pingDelay = pingServer(callBaseUri, "", 0); int pingRetries = 0; int maxPingRetries = 10; // Prevent infinite loop diff --git a/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingPartRequestTest.java b/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingPartRequestTest.java new file mode 100644 index 0000000000..b1fa774dbe --- /dev/null +++ b/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingPartRequestTest.java @@ -0,0 +1,94 @@ +/* + * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved. + */ +package com.marklogic.client.impl; + +import com.marklogic.client.DatabaseClient; +import com.marklogic.client.DatabaseClientFactory; +import com.marklogic.client.MarkLogicIOException; +import mockwebserver3.MockResponse; +import mockwebserver3.MockWebServer; +import mockwebserver3.RecordedRequest; +import mockwebserver3.SocketEffect; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.StringReader; +import java.util.stream.Stream; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Uses a mock server to verify how a Data Services call with a non-resendable streaming part (e.g. a Reader) is sent. + * A real MarkLogic instance can close a pooled connection after a response, such as a compressed rows response. The + * next streaming request would then fail on the dead connection, and OkHttp would try to resend the already-consumed + * stream, resulting in a misleading "Stream closed" error. + */ +class StreamingPartRequestTest { + + private MockWebServer mockWebServer; + private DatabaseClient client; + + @BeforeEach + void beforeEach() throws IOException { + mockWebServer = new MockWebServer(); + mockWebServer.start(); + client = DatabaseClientFactory.newClient("localhost", mockWebServer.getPort(), + new DatabaseClientFactory.BasicAuthContext("user", "password")); + } + + @AfterEach + void afterEach() { + client.release(); + mockWebServer.close(); + } + + @Test + void serverIsPingedBeforeEveryStreamingRequest() throws Exception { + for (int i = 0; i < 2; i++) { + mockWebServer.enqueue(new MockResponse.Builder().code(200).build()); + mockWebServer.enqueue(new MockResponse.Builder().code(200).build()); + } + + callWithStreamingPart(); + callWithStreamingPart(); + + assertEquals("HEAD", mockWebServer.takeRequest().getMethod()); + assertEquals("POST", mockWebServer.takeRequest().getMethod()); + assertEquals("HEAD", mockWebServer.takeRequest().getMethod(), + "The ping must not be limited to the first request, as a connection used by an earlier request may " + + "have been closed by the server."); + RecordedRequest secondPost = mockWebServer.takeRequest(); + assertEquals("POST", secondPost.getMethod()); + assertTrue(secondPost.getBody().utf8().contains("{\"a\":1}")); + } + + @Test + void failedStreamingRequestIsNotResent() { + mockWebServer.enqueue(new MockResponse.Builder().code(200).build()); + mockWebServer.enqueue(new MockResponse.Builder() + .onResponseStart(new SocketEffect.CloseSocket()) + .build()); + // Would be consumed if the streaming request were resent. + mockWebServer.enqueue(new MockResponse.Builder().code(200).build()); + + MarkLogicIOException ex = assertThrows(MarkLogicIOException.class, this::callWithStreamingPart); + + assertFalse(ex.getMessage().contains("Stream closed"), + "The stream cannot be resent, so the original failure should be reported instead of an attempt to " + + "read the consumed stream; message: " + ex.getMessage()); + assertTrue(ex.getMessage().contains("unexpected end of stream"), "Unexpected message: " + ex.getMessage()); + assertEquals(2, mockWebServer.getRequestCount()); + } + + private void callWithStreamingPart() { + BaseProxy.request("/ds/custom-content/", "importContent.xqy", BaseProxy.ParameterValuesKind.MULTIPLE_NODES) + .on(client) + .withSession() + .withParams(BaseProxy.documentParam("content", false, + Stream.of(new StringReader("{\"a\":1}")).map(BaseProxy.JsonDocumentType::fromReader))) + .responseNone(); + } +}