MLE-32970 Fix multipart streaming input to data services

Author: Ryan Dew

marklogic-client-api/src/main/java/com/marklogic/client/impl/OkHttpServices.java

@@ -5194,7 +5194,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;
@@ -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

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

@@ -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;
+	}
 }

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

@@ -6,6 +6,7 @@
 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);
 		}
 

marklogic-client-api/src/test/java/com/marklogic/client/impl/StreamingOutputImplTest.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright (c) 2010-2025 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());
+	}
+}