Skip to content

Commit 2c71fdb

Browse files
authored
Merge pull request #175 from conductor-oss/feat/no-transparent-connection-retry
feat(client): add retransmitRequestBodies(false) to stop re-sending a request that was already sent
2 parents 2055512 + 759a8d3 commit 2c71fdb

2 files changed

Lines changed: 227 additions & 2 deletions

File tree

‎conductor-client/src/main/java/com/netflix/conductor/client/http/ConductorClient.java‎

Lines changed: 55 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@
6868
import okhttp3.RequestBody;
6969
import okhttp3.Response;
7070
import okhttp3.internal.http.HttpMethod;
71+
import okio.BufferedSink;
7172

7273
public class ConductorClient {
7374
private static final Logger LOGGER = LoggerFactory.getLogger(ConductorClient.class);
@@ -79,6 +80,7 @@ public class ConductorClient {
7980
private final KeyManager[] keyManagers;
8081
private final List<HeaderSupplier> headerSuppliers;
8182
private final MetricsCollector metricsCollector;
83+
private final boolean retransmitRequestBodies;
8284

8385
public static Builder<?> builder() {
8486
return new Builder<>();
@@ -95,6 +97,7 @@ protected ConductorClient(Builder<?> builder) {
9597
this.keyManagers = builder.keyManagers;
9698
this.headerSuppliers = builder.headerSupplier();
9799
this.metricsCollector = builder.metricsCollector;
100+
this.retransmitRequestBodies = builder.retransmitRequestBodies;
98101

99102
if (this.metricsCollector != null) {
100103
ApiClientMetrics apiClientMetrics = this.metricsCollector.getApiClientMetrics();
@@ -491,13 +494,47 @@ private RequestBody requestBody(String method, String contentType, Object body)
491494
return null;
492495
}
493496

497+
RequestBody requestBody;
494498
if (body == null && "DELETE".equals(method)) {
495499
return null;
496500
} else if (body == null) {
497-
return RequestBody.create("", MediaType.parse(contentType));
501+
requestBody = RequestBody.create("", MediaType.parse(contentType));
502+
} else {
503+
requestBody = serialize(contentType, body);
498504
}
499505

500-
return serialize(contentType, body);
506+
return retransmitRequestBodies ? requestBody : oneShot(requestBody);
507+
}
508+
509+
// Wraps a request body so OkHttp will not retransmit it on a retried connection.
510+
private static RequestBody oneShot(RequestBody delegate) {
511+
return new RequestBody() {
512+
@Override
513+
public MediaType contentType() {
514+
return delegate.contentType();
515+
}
516+
517+
@Override
518+
public long contentLength() throws IOException {
519+
return delegate.contentLength();
520+
}
521+
522+
@Override
523+
public void writeTo(@NotNull BufferedSink sink) throws IOException {
524+
delegate.writeTo(sink);
525+
}
526+
527+
@Override
528+
public boolean isDuplex() {
529+
return delegate.isDuplex();
530+
}
531+
532+
// isOneShot() == true means: never re-send a body that was already transmitted.
533+
@Override
534+
public boolean isOneShot() {
535+
return true;
536+
}
537+
};
501538
}
502539

503540
private HttpUrl buildUrl(String path, List<Param> queryParams) {
@@ -603,6 +640,7 @@ public static class Builder<T extends Builder<T>> {
603640
private Supplier<ObjectMapper> objectMapperSupplier = () -> new ObjectMapperProvider().getObjectMapper();
604641
private final List<HeaderSupplier> headerSuppliers = new ArrayList<>();
605642
MetricsCollector metricsCollector;
643+
private boolean retransmitRequestBodies = true;
606644

607645
private boolean useEnvVariables = false;
608646

@@ -656,6 +694,21 @@ public T proxy(Proxy proxy) {
656694
return self();
657695
}
658696

697+
/**
698+
* Pass {@code false} to mark request bodies one-shot, opting in to the protection: once
699+
* transmission of a body has begun, it is never sent again on a retried connection. This
700+
* also stops OkHttp replaying the request on 307/308 redirects, 408 responses and 421
701+
* misdirected-request responses (301/302/303 are unaffected, since those convert to a
702+
* bodyless GET). A blocked 307/308 surfaces as a {@link ConductorClientException}
703+
* carrying that status code rather than a transparent redirect, since
704+
* {@link ConductorClient#handleResponse} treats 3xx as unsuccessful. Defaults to
705+
* {@code true}, OkHttp's stock retransmit/replay behaviour.
706+
*/
707+
public T retransmitRequestBodies(boolean retransmitRequestBodies) {
708+
this.retransmitRequestBodies = retransmitRequestBodies;
709+
return self();
710+
}
711+
659712
public T connectionPoolConfig(ConnectionPoolConfig config) {
660713
this.connectionPoolConfig = config;
661714
return self();
Lines changed: 172 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,172 @@
1+
/*
2+
* Copyright 2026 Conductor Authors.
3+
* <p>
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
5+
* the License. You may obtain a copy of the License at
6+
* <p>
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
* <p>
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
10+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
11+
* specific language governing permissions and limitations under the License.
12+
*/
13+
package com.netflix.conductor.client.http;
14+
15+
import java.io.IOException;
16+
import java.io.InterruptedIOException;
17+
import java.net.InetAddress;
18+
import java.util.ArrayList;
19+
import java.util.List;
20+
import java.util.Map;
21+
22+
import org.junit.jupiter.api.AfterEach;
23+
import org.junit.jupiter.api.BeforeEach;
24+
import org.junit.jupiter.api.DisplayName;
25+
import org.junit.jupiter.api.Test;
26+
27+
import com.netflix.conductor.client.exception.ConductorClientException;
28+
29+
import okhttp3.Dns;
30+
import okhttp3.RequestBody;
31+
import okhttp3.mockwebserver.MockResponse;
32+
import okhttp3.mockwebserver.MockWebServer;
33+
import okhttp3.mockwebserver.SocketPolicy;
34+
35+
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
36+
import static org.junit.jupiter.api.Assertions.assertEquals;
37+
import static org.junit.jupiter.api.Assertions.assertFalse;
38+
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
39+
import static org.junit.jupiter.api.Assertions.assertThrows;
40+
import static org.junit.jupiter.api.Assertions.assertTrue;
41+
42+
/**
43+
* A request body that is not one-shot can be sent twice: OkHttp retransmits it after a
44+
* recoverable connection failure, and the caller is then told about the second attempt rather
45+
* than the first. For a non-idempotent call that means the work happened and the error says it
46+
* did not. {@code retransmitRequestBodies(false)} opts in to the one-shot protection.
47+
*
48+
* <p>The decisive pair below makes one successful call first, so the connection is pooled, then
49+
* severs the next one: a connection taken from the pool retries on its own address, which is the
50+
* real-world shape this SDK must handle. The remaining tests pin the one-shot contract directly
51+
* and check that a failure before the request is sent still falls back to another route - the
52+
* behaviour that must not be lost in exchange.
53+
*/
54+
class RequestBodyRetransmissionTest {
55+
56+
private MockWebServer server;
57+
private final List<ConductorClient> clients = new ArrayList<>();
58+
59+
@BeforeEach
60+
void setUp() throws IOException {
61+
server = new MockWebServer();
62+
server.start(InetAddress.getByName("127.0.0.1"), 0);
63+
}
64+
65+
@AfterEach
66+
void tearDown() throws IOException {
67+
// Pools are per-client here, but evict explicitly so no pooled connection outlives its server.
68+
clients.forEach(c -> c.okHttpClient.connectionPool().evictAll());
69+
server.shutdown();
70+
}
71+
72+
@Test
73+
@DisplayName("default (retransmitRequestBodies true): a body severed on a pooled connection "
74+
+ "is retransmitted and the call succeeds")
75+
void retransmitEnabled_bodyIsRetransmittedOnPooledConnection_callSucceeds() {
76+
server.enqueue(new MockResponse().setBody("{}")); // warm-up: pools the connection
77+
server.enqueue(new MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AFTER_REQUEST));
78+
server.enqueue(new MockResponse().setBody("{}")); // served only if retransmitted
79+
80+
var client = newClient(ConductorClient.builder().basePath(basePath()).retransmitRequestBodies(true));
81+
82+
// Must run to completion (response body drained) or the connection is never pooled and the retry never fires.
83+
assertDoesNotThrow(() -> client.execute(warmupRequest()));
84+
85+
assertDoesNotThrow(() -> client.execute(postRequest()),
86+
"retransmission must hide the severed attempt from the caller");
87+
assertEquals(3, server.getRequestCount(), "warm-up + severed attempt + retransmit");
88+
}
89+
90+
@Test
91+
@DisplayName("opt-in (retransmitRequestBodies(false)): a body severed on a pooled connection "
92+
+ "is not retransmitted and the call fails")
93+
void optIn_bodyIsNotRetransmittedOnPooledConnection_callFails() {
94+
server.enqueue(new MockResponse().setBody("{}")); // warm-up: pools the connection
95+
server.enqueue(new MockResponse().setSocketPolicy(SocketPolicy.DISCONNECT_AFTER_REQUEST));
96+
server.enqueue(new MockResponse().setBody("{}")); // must never be reached
97+
98+
var client = newClient(ConductorClient.builder().basePath(basePath()).retransmitRequestBodies(false));
99+
100+
// Must run to completion (response body drained) or the connection is never pooled and the retry never fires.
101+
assertDoesNotThrow(() -> client.execute(warmupRequest()));
102+
103+
ConductorClientException e = assertThrows(ConductorClientException.class, () -> client.execute(postRequest()));
104+
105+
assertEquals(2, server.getRequestCount(), "warm-up + severed attempt, no retransmit");
106+
assertInstanceOf(IOException.class, e.getCause(), "got: " + e.getCause());
107+
assertFalse(e.getCause() instanceof InterruptedIOException,
108+
"must be the raw severed-socket failure, not a call timeout masquerading as it: " + e.getCause());
109+
}
110+
111+
@Test
112+
@DisplayName("contract pin: the built request body is one-shot only when opted in "
113+
+ "(retransmitRequestBodies(false))")
114+
void requestBody_isOneShot_onlyWhenOptedIn() {
115+
var defaultClient = newClient(ConductorClient.builder().basePath(basePath()));
116+
var optInClient = newClient(ConductorClient.builder().basePath(basePath()).retransmitRequestBodies(false));
117+
118+
assertFalse(builtRequestBody(defaultClient).isOneShot());
119+
assertTrue(builtRequestBody(optInClient).isOneShot());
120+
}
121+
122+
@Test
123+
@DisplayName("a pre-send connection failure still falls back to another route, even for a "
124+
+ "one-shot POST body")
125+
void preSendConnectFailure_fallsBackToAnotherRoute() {
126+
server.enqueue(new MockResponse().setBody("{}"));
127+
128+
Dns twoRouteDns = hostname -> List.of(
129+
InetAddress.getByName("127.0.0.2"), InetAddress.getByName("127.0.0.1"));
130+
131+
var client = newClient(
132+
ConductorClient.builder()
133+
.basePath("http://multi-route.invalid:" + server.getPort() + "/api")
134+
.retransmitRequestBodies(false)
135+
.connectTimeout(250)
136+
.configureOkHttp(b -> b.dns(twoRouteDns)));
137+
138+
assertDoesNotThrow(() -> client.execute(postRequest()),
139+
"a one-shot POST must still fall back pre-send: recover() never consults "
140+
+ "requestIsOneShot before the body starts sending");
141+
assertEquals(1, server.getRequestCount(), "request must have reached the server via the fallback route");
142+
}
143+
144+
private ConductorClient newClient(ConductorClient.Builder<?> builder) {
145+
ConductorClient client = new ConductorClient(builder);
146+
clients.add(client);
147+
return client;
148+
}
149+
150+
private String basePath() {
151+
return "http://127.0.0.1:" + server.getPort() + "/api";
152+
}
153+
154+
private static ConductorClientRequest warmupRequest() {
155+
return ConductorClientRequest.builder()
156+
.method(ConductorClientRequest.Method.GET)
157+
.path("/workflow")
158+
.build();
159+
}
160+
161+
private static ConductorClientRequest postRequest() {
162+
return ConductorClientRequest.builder()
163+
.method(ConductorClientRequest.Method.POST)
164+
.path("/workflow")
165+
.body("{\"name\":\"test\"}")
166+
.build();
167+
}
168+
169+
private static RequestBody builtRequestBody(ConductorClient client) {
170+
return client.buildRequest("POST", "/workflow", List.of(), List.of(), Map.of(), "{\"name\":\"test\"}").body();
171+
}
172+
}

0 commit comments

Comments
 (0)