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 1de3b0bff..257d8c81d 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)); @@ -5194,7 +5205,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 +5247,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 @@ -5282,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); } @@ -5504,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 5fd30e6d0..62cdf697d 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; @@ -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,7 +65,19 @@ 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 + 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 !resendable; } } 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 b2f57e0c3..d2517828c 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 000000000..f92155899 --- /dev/null +++ b/marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java @@ -0,0 +1,146 @@ +/* + * 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.assertFalse; +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()); + } + + @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."); + } +}