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..4f6db8242 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); } @@ -5389,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 @@ -5504,7 +5529,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."); + } +} 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 000000000..b1fa774db --- /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(); + } +}