From 3c26e656cb140f3a3ef3255b6345a5c02a9ca163 Mon Sep 17 00:00:00 2001 From: Ryan Dew Date: Tue, 22 Sep 2026 19:01:34 -0400 Subject: [PATCH 1/2] 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/2] 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."); + } }