Skip to content

Commit af0a197

Browse files
author
Ryan Dew
committed
MLE-32953 Fix multipart streaming input to data services
1 parent 1d19510 commit af0a197

3 files changed

Lines changed: 65 additions & 1 deletion

File tree

‎marklogic-client-api/src/main/java/com/marklogic/client/impl/StreamingOutputImpl.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,4 +53,13 @@ public boolean isRetryable() {
5353
// Added in 8.0.0; streaming output cannot be retried as the stream is consumed on first write.
5454
return false;
5555
}
56+
57+
@Override
58+
public boolean isOneShot() {
59+
// Declares this body as one-shot via OkHttp's own contract so that when this body is nested inside another
60+
// RequestBody (e.g. a MultipartBody part for a Data Services call), OkHttp and RetryIOExceptionInterceptor
61+
// both recognize the outer body as non-retryable too; isRetryable() above is only checked when this is the
62+
// top-level request body and would otherwise be bypassed for a nested streaming part.
63+
return true;
64+
}
5665
}

‎marklogic-client-api/src/main/java/com/marklogic/client/impl/okhttp/RetryIOExceptionInterceptor.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import com.marklogic.client.MarkLogicIOException;
77
import okhttp3.Interceptor;
88
import okhttp3.Request;
9+
import okhttp3.RequestBody;
910
import okhttp3.Response;
1011
import org.slf4j.Logger;
1112

@@ -40,8 +41,10 @@ public RetryIOExceptionInterceptor(int maxRetries, long initialDelayMs, double b
4041
@Override
4142
public Response intercept(Chain chain) throws IOException {
4243
Request request = chain.request();
44+
RequestBody requestBody = request.body();
4345

44-
if (request.body() instanceof RetryableRequestBody body && !body.isRetryable()) {
46+
if (requestBody != null && (requestBody.isOneShot() ||
47+
requestBody instanceof RetryableRequestBody body && !body.isRetryable())) {
4548
return chain.proceed(request);
4649
}
4750

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
3+
*/
4+
package com.marklogic.client.impl;
5+
6+
import com.marklogic.client.impl.okhttp.RetryIOExceptionInterceptor;
7+
import okhttp3.MediaType;
8+
import okhttp3.MultipartBody;
9+
import okhttp3.OkHttpClient;
10+
import okhttp3.Request;
11+
import org.junit.jupiter.api.Test;
12+
13+
import java.io.IOException;
14+
import java.nio.charset.StandardCharsets;
15+
import java.util.concurrent.atomic.AtomicInteger;
16+
17+
import static org.junit.jupiter.api.Assertions.assertEquals;
18+
import static org.junit.jupiter.api.Assertions.assertThrows;
19+
import static org.junit.jupiter.api.Assertions.assertTrue;
20+
21+
class StreamingOutputImplTest {
22+
23+
@Test
24+
void multipartBodyWithStreamingPartIsNotRetried() {
25+
StreamingOutputImpl streamingBody = new StreamingOutputImpl(
26+
outputStream -> outputStream.write("content".getBytes(StandardCharsets.UTF_8)),
27+
null,
28+
MediaType.get("application/xml")
29+
);
30+
MultipartBody multipartBody = new MultipartBody.Builder()
31+
.setType(MultipartBody.FORM)
32+
.addFormDataPart("body", null, streamingBody)
33+
.build();
34+
assertTrue(multipartBody.isOneShot());
35+
36+
AtomicInteger attempts = new AtomicInteger();
37+
OkHttpClient client = new OkHttpClient.Builder()
38+
.addInterceptor(new RetryIOExceptionInterceptor(3, 0, 1, 0))
39+
.addInterceptor(chain -> {
40+
attempts.incrementAndGet();
41+
throw new IOException("unexpected end of stream");
42+
})
43+
.build();
44+
Request request = new Request.Builder()
45+
.url("http://localhost/")
46+
.post(multipartBody)
47+
.build();
48+
49+
assertThrows(IOException.class, () -> client.newCall(request).execute());
50+
assertEquals(1, attempts.get());
51+
}
52+
}

0 commit comments

Comments
 (0)