Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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) :
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -2831,7 +2831,7 @@ public <R extends AbstractReadHandle> R putResource(RequestLogger reqlog,

Function<Request.Builder, Response> doPutFunction = new Function<Request.Builder, Response>() {
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);
Expand Down Expand Up @@ -2980,7 +2980,7 @@ public void accept(Boolean resendable) {
final Object value = inputBase == null ? null : inputBase.sendContent();
Function<Request.Builder, Response> doPostFunction = new Function<Request.Builder, Response>() {
public Response apply(Request.Builder funcBuilder) {
return doPost(reqlog, funcBuilder, value);
return doPost(reqlog, funcBuilder, value, isResendable);
}
};

Expand Down Expand Up @@ -3573,7 +3573,8 @@ private <U extends OkHttpResultIterator> U postIteratedResourceImpl(
Function<Request.Builder, Response> doPostFunction = requestBuilder -> doPost(
reqlog,
requestBuilder.header(HEADER_ACCEPT, multipartMixedWithBoundary()),
inputBase.sendContent()
inputBase.sendContent(),
isResendable
);

Response response = sendRequestWithRetry(requestBldr, (transaction == null), doPostFunction, resendableConsumer);
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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);
}
Expand All @@ -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));
Expand Down Expand Up @@ -4003,7 +4014,7 @@ private <W extends AbstractWriteHandle> 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));
Expand Down Expand Up @@ -5194,7 +5205,7 @@ static private List<BodyPart> getPartList(MimeMultipart multipart) {
}
}

static private class ObjectRequestBody extends RequestBody implements RetryableRequestBody {
static class ObjectRequestBody extends RequestBody implements RetryableRequestBody {

private Object obj;
private MediaType contentType;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
}

Expand Down Expand Up @@ -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);
}

Expand Down
Original file line number Diff line number Diff line change
@@ -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;

Expand All @@ -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
Expand All @@ -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;
}
}
Original file line number Diff line number Diff line change
@@ -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;

Expand Down Expand Up @@ -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())) {
Comment thread
rjdew-progress marked this conversation as resolved.
return chain.proceed(request);
}

Expand Down
Original file line number Diff line number Diff line change
@@ -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.");
}
}
Loading