Skip to content

Commit c314875

Browse files
skyflow-bhartiskyflow-bharti
committed
SK-3026 added retries
1 parent 5bbe58c commit c314875

5 files changed

Lines changed: 968 additions & 3 deletions

File tree

‎flowvault/src/main/java/com/skyflow/Skyflow.java‎

Lines changed: 125 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,15 @@ public VaultController vault() throws SkyflowException {
4646

4747
public static final class SkyflowClientBuilder extends BaseSkyflowClientBuilder<VaultConfig> {
4848
private final LinkedHashMap<String, VaultController> vaultClientsMap = new LinkedHashMap<>();
49+
// Client-wide HTTP config. Resolution per vault, most specific first:
50+
// VaultConfig value -> the value set here -> SDK default (60s call timeout, 0 retries).
51+
// null here means "not set", so the SDK default applies to vaults that don't override it.
52+
// Only null means inherit: an explicit 0 is a real value and wins over the level below.
53+
private Integer timeout;
54+
private Integer connectTimeout;
55+
private Integer readTimeout;
56+
private Integer writeTimeout;
57+
private Integer maxRetries;
4958

5059
@Override
5160
protected void validateVaultConfig(VaultConfig vaultConfig) throws SkyflowException {
@@ -54,13 +63,19 @@ protected void validateVaultConfig(VaultConfig vaultConfig) throws SkyflowExcept
5463

5564
@Override
5665
protected void onVaultConfigAdded(VaultConfig vaultConfig) throws SkyflowException {
57-
this.vaultClientsMap.put(vaultConfig.getVaultId(), new VaultController(vaultConfig, this.skyflowCredentials));
66+
VaultController controller = new VaultController(vaultConfig, this.skyflowCredentials);
67+
controller.setCommonHttpConfig(this.timeout, this.connectTimeout, this.readTimeout,
68+
this.writeTimeout, this.maxRetries);
69+
this.vaultClientsMap.put(vaultConfig.getVaultId(), controller);
5870
LogUtil.printInfoLog(Utils.parameterizedString(InfoLogs.VAULT_CONTROLLER_INITIALIZED.getLog(), vaultConfig.getVaultId()));
5971
}
6072

6173
@Override
6274
protected void onVaultConfigUpdated(VaultConfig updatedConfig) throws SkyflowException {
63-
this.vaultClientsMap.put(updatedConfig.getVaultId(), new VaultController(updatedConfig, this.skyflowCredentials));
75+
VaultController updated = new VaultController(updatedConfig, this.skyflowCredentials);
76+
updated.setCommonHttpConfig(this.timeout, this.connectTimeout, this.readTimeout,
77+
this.writeTimeout, this.maxRetries);
78+
this.vaultClientsMap.put(updatedConfig.getVaultId(), updated);
6479
}
6580

6681
@Override
@@ -89,9 +104,48 @@ public SkyflowClientBuilder addVaultConfig(VaultConfig vaultConfig) throws Skyfl
89104
@Override
90105
public SkyflowClientBuilder updateVaultConfig(VaultConfig vaultConfig) throws SkyflowException {
91106
super.updateVaultConfig(vaultConfig);
107+
carryVaultOverrides(vaultConfig);
92108
return this;
93109
}
94110

111+
/**
112+
* BaseSkyflow.mergeVaultConfig() only carries env, clusterId and credentials across, so the
113+
* flowvault-specific fields on an incoming update — vaultURL and the HTTP settings — would
114+
* be dropped silently. Apply them to the merged config the new controller is holding. A null
115+
* on the incoming config means "leave as is", matching how the base class merges every
116+
* other field.
117+
*/
118+
private void carryVaultOverrides(VaultConfig incoming) throws SkyflowException {
119+
VaultConfig merged = this.vaultConfigMap.get(incoming.getVaultId());
120+
if (merged == null || merged == incoming) {
121+
return;
122+
}
123+
if (incoming.getTimeout() != null) {
124+
merged.setTimeout(incoming.getTimeout());
125+
}
126+
if (incoming.getConnectTimeout() != null) {
127+
merged.setConnectTimeout(incoming.getConnectTimeout());
128+
}
129+
if (incoming.getReadTimeout() != null) {
130+
merged.setReadTimeout(incoming.getReadTimeout());
131+
}
132+
if (incoming.getWriteTimeout() != null) {
133+
merged.setWriteTimeout(incoming.getWriteTimeout());
134+
}
135+
if (incoming.getMaxRetries() != null) {
136+
merged.setMaxRetries(incoming.getMaxRetries());
137+
}
138+
// The HTTP settings above are resolved lazily on the next request, but the URL is
139+
// resolved once in the VaultClient constructor — which already ran with the old value.
140+
if (incoming.getVaultURL() != null) {
141+
merged.setVaultURL(incoming.getVaultURL());
142+
VaultController controller = this.vaultClientsMap.get(incoming.getVaultId());
143+
if (controller != null) {
144+
controller.refreshVaultURL();
145+
}
146+
}
147+
}
148+
95149
@Override
96150
public SkyflowClientBuilder removeVaultConfig(String vaultId) throws SkyflowException {
97151
super.removeVaultConfig(vaultId);
@@ -110,6 +164,75 @@ public SkyflowClientBuilder setLogLevel(LogLevel logLevel) {
110164
return this;
111165
}
112166

167+
/**
168+
* Overall call timeout in seconds, including retries. Default 60.
169+
* <p>
170+
* <b>Precedence:</b> a vault that sets {@link VaultConfig#setTimeout(Integer)} wins; this
171+
* value applies only to vaults that leave it unset.
172+
*/
173+
public SkyflowClientBuilder timeout(int timeout) {
174+
this.timeout = timeout;
175+
propagateHttpConfig();
176+
return this;
177+
}
178+
179+
/**
180+
* Per-attempt connection-establishment timeout in seconds. Unset => HTTP client default (10s).
181+
* <p>
182+
* <b>Precedence:</b> a vault that sets {@link VaultConfig#setConnectTimeout(Integer)} wins;
183+
* this value applies only to vaults that leave it unset.
184+
*/
185+
public SkyflowClientBuilder connectTimeout(int connectTimeout) {
186+
this.connectTimeout = connectTimeout;
187+
propagateHttpConfig();
188+
return this;
189+
}
190+
191+
/**
192+
* Per-attempt response-read timeout in seconds. Unset => HTTP client default (10s).
193+
* <p>
194+
* <b>Precedence:</b> a vault that sets {@link VaultConfig#setReadTimeout(Integer)} wins;
195+
* this value applies only to vaults that leave it unset.
196+
*/
197+
public SkyflowClientBuilder readTimeout(int readTimeout) {
198+
this.readTimeout = readTimeout;
199+
propagateHttpConfig();
200+
return this;
201+
}
202+
203+
/**
204+
* Per-attempt request-write timeout in seconds. Unset => HTTP client default (10s).
205+
* <p>
206+
* <b>Precedence:</b> a vault that sets {@link VaultConfig#setWriteTimeout(Integer)} wins;
207+
* this value applies only to vaults that leave it unset.
208+
*/
209+
public SkyflowClientBuilder writeTimeout(int writeTimeout) {
210+
this.writeTimeout = writeTimeout;
211+
propagateHttpConfig();
212+
return this;
213+
}
214+
215+
/**
216+
* Retry attempts after the first failure. Default 0 — retries are opt-in so non-idempotent
217+
* bulk writes are not replayed automatically.
218+
* <p>
219+
* <b>Precedence:</b> a vault that sets {@link VaultConfig#setMaxRetries(Integer)} wins;
220+
* this value applies only to vaults that leave it unset.
221+
*/
222+
public SkyflowClientBuilder maxRetries(int maxRetries) {
223+
this.maxRetries = maxRetries;
224+
propagateHttpConfig();
225+
return this;
226+
}
227+
228+
/** Push the current client-wide HTTP settings onto every vault controller built so far. */
229+
private void propagateHttpConfig() {
230+
for (VaultController vault : this.vaultClientsMap.values()) {
231+
vault.setCommonHttpConfig(this.timeout, this.connectTimeout, this.readTimeout,
232+
this.writeTimeout, this.maxRetries);
233+
}
234+
}
235+
113236
public Skyflow build() {
114237
return new Skyflow(this);
115238
}

‎flowvault/src/main/java/com/skyflow/VaultClient.java‎

Lines changed: 86 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,30 @@
55
import com.skyflow.errors.SkyflowException;
66
import com.skyflow.generated.rest.ApiClient;
77
import com.skyflow.generated.rest.ApiClientBuilder;
8+
import com.skyflow.generated.rest.core.RetryInterceptor;
89
import com.skyflow.generated.rest.resources.flowservice.FlowserviceClient;
910
import com.skyflow.generated.rest.resources.records.RecordsClient;
1011
import com.skyflow.utils.Utils;
1112

13+
import java.util.concurrent.TimeUnit;
14+
15+
import okhttp3.ConnectionPool;
16+
import okhttp3.OkHttpClient;
17+
import okhttp3.Request;
18+
1219
public class VaultClient extends BaseVaultClient<VaultConfig> {
1320
private final ApiClientBuilder apiClientBuilder;
1421
private ApiClient apiClient;
22+
// Client-wide (Skyflow builder) HTTP config; null => fall back to the SDK defaults below.
23+
private Integer commonTimeout;
24+
private Integer commonConnectTimeout;
25+
private Integer commonReadTimeout;
26+
private Integer commonWriteTimeout;
27+
private Integer commonMaxRetries;
28+
// SDK defaults, used when neither the vault-level nor the client-wide value is set.
29+
private static final int DEFAULT_TIMEOUT_SECONDS = 60;
30+
// Retries OFF by default (opt-in) so non-idempotent bulk writes aren't replayed automatically.
31+
private static final int DEFAULT_MAX_RETRIES = 0;
1532

1633
protected VaultClient(VaultConfig vaultConfig, Credentials credentials) throws SkyflowException {
1734
super(vaultConfig, credentials);
@@ -20,6 +37,37 @@ protected VaultClient(VaultConfig vaultConfig, Credentials credentials) throws S
2037
updateVaultURL();
2138
}
2239

40+
/**
41+
* Applies the client-wide HTTP settings from the Skyflow builder. Discards the cached HTTP
42+
* client and ApiClient so the next call rebuilds them with the new values.
43+
*/
44+
protected void setCommonHttpConfig(Integer timeout, Integer connectTimeout, Integer readTimeout,
45+
Integer writeTimeout, Integer maxRetries) {
46+
this.commonTimeout = timeout;
47+
this.commonConnectTimeout = connectTimeout;
48+
this.commonReadTimeout = readTimeout;
49+
this.commonWriteTimeout = writeTimeout;
50+
this.commonMaxRetries = maxRetries;
51+
this.sharedHttpClient = null;
52+
this.apiClient = null;
53+
}
54+
55+
/** Resolve a setting: vault-level override, else client-wide default, else the SDK default. */
56+
private static int resolveInt(Integer vaultLevel, Integer clientLevel, int defaultValue) {
57+
if (vaultLevel != null) {
58+
return vaultLevel;
59+
}
60+
return clientLevel != null ? clientLevel : defaultValue;
61+
}
62+
63+
/**
64+
* Resolve an optional setting: vault-level override, else client-wide default, else null.
65+
* Null means "not configured" — the caller leaves the underlying HTTP client default in place.
66+
*/
67+
private static Integer resolveNullableInt(Integer vaultLevel, Integer clientLevel) {
68+
return vaultLevel != null ? vaultLevel : clientLevel;
69+
}
70+
2371
protected FlowserviceClient getRecordsApi() {
2472
return this.apiClient.flowservice();
2573
}
@@ -41,6 +89,14 @@ protected synchronized void setBearerToken() throws SkyflowException {
4189
}
4290
}
4391

92+
/**
93+
* Re-resolves the vault URL from the current config. The constructor resolves it once, so a
94+
* vaultURL supplied later through updateVaultConfig would otherwise never take effect.
95+
*/
96+
protected void refreshVaultURL() throws SkyflowException {
97+
updateVaultURL();
98+
}
99+
44100
private void updateVaultURL() throws SkyflowException {
45101
// Fetch vaultURL from ENV
46102
String vaultURL = Utils.getEnvVaultURL();
@@ -63,7 +119,36 @@ private void updateVaultURL() throws SkyflowException {
63119

64120
protected void updateExecutorInHTTP() {
65121
if (sharedHttpClient == null) {
66-
sharedHttpClient = buildSharedHttpClient(() -> this.token);
122+
int timeoutSeconds = resolveInt(vaultConfig.getTimeout(), commonTimeout, DEFAULT_TIMEOUT_SECONDS);
123+
int maxRetries = resolveInt(vaultConfig.getMaxRetries(), commonMaxRetries, DEFAULT_MAX_RETRIES);
124+
// Per-attempt timeouts: null => leave OkHttp's built-in default (backward compatible).
125+
Integer connectTimeout = resolveNullableInt(vaultConfig.getConnectTimeout(), commonConnectTimeout);
126+
Integer readTimeout = resolveNullableInt(vaultConfig.getReadTimeout(), commonReadTimeout);
127+
Integer writeTimeout = resolveNullableInt(vaultConfig.getWriteTimeout(), commonWriteTimeout);
128+
129+
OkHttpClient.Builder httpBuilder = new OkHttpClient.Builder()
130+
.connectionPool(new ConnectionPool(10, 1, TimeUnit.MINUTES))
131+
// Overall ceiling; bounds the whole call including retries.
132+
.callTimeout(timeoutSeconds, TimeUnit.SECONDS)
133+
// OUTER: retries. Must wrap the auth interceptor so each attempt re-reads the
134+
// (possibly refreshed) bearer token rather than replaying a stale one.
135+
.addInterceptor(new RetryInterceptor(maxRetries))
136+
.addInterceptor(chain -> { // INNER: auth
137+
Request requestWithAuth = chain.request().newBuilder()
138+
.header("Authorization", "Bearer " + this.token)
139+
.build();
140+
return chain.proceed(requestWithAuth);
141+
});
142+
if (connectTimeout != null) {
143+
httpBuilder.connectTimeout(connectTimeout, TimeUnit.SECONDS);
144+
}
145+
if (readTimeout != null) {
146+
httpBuilder.readTimeout(readTimeout, TimeUnit.SECONDS);
147+
}
148+
if (writeTimeout != null) {
149+
httpBuilder.writeTimeout(writeTimeout, TimeUnit.SECONDS);
150+
}
151+
sharedHttpClient = httpBuilder.build();
67152
apiClientBuilder.httpClient(sharedHttpClient);
68153
}
69154
}

‎flowvault/src/main/java/com/skyflow/config/VaultConfig.java‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,35 @@
11
package com.skyflow.config;
22

3+
/**
4+
* Per-vault configuration.
5+
* <p>
6+
* The HTTP timeout and retry settings below are <b>vault-level overrides</b>. Each one resolves
7+
* most-specific-first: the value set here, else the client-wide value set on
8+
* {@code Skyflow.builder()}, else the SDK default. So when the same setting is supplied at both
9+
* levels, <b>the value on this VaultConfig takes precedence</b> and the client-wide value is
10+
* ignored for this vault.
11+
* <p>
12+
* Only {@code null} means "inherit" — an explicit {@code 0} is a real value and wins over the
13+
* client-wide setting.
14+
*/
315
public class VaultConfig extends BaseVaultConfig {
416

517
private String vaultURL;
18+
// HTTP timeout & retry config (vault-level overrides). null => inherit client-wide default, then SDK default.
19+
private Integer timeout; // overall call timeout, in seconds
20+
private Integer connectTimeout; // per-attempt connection-establishment timeout, in seconds
21+
private Integer readTimeout; // per-attempt response-read timeout, in seconds
22+
private Integer writeTimeout; // per-attempt request-write timeout, in seconds
23+
private Integer maxRetries; // retry attempts after the first failure
624

725
public VaultConfig() {
826
super();
927
this.vaultURL = null;
28+
this.timeout = null;
29+
this.connectTimeout = null;
30+
this.readTimeout = null;
31+
this.writeTimeout = null;
32+
this.maxRetries = null;
1033
}
1134

1235
public String getVaultURL() {
@@ -17,4 +40,78 @@ public void setVaultURL(String vaultURL) {
1740
this.vaultURL = vaultURL;
1841
}
1942

43+
public Integer getTimeout() {
44+
return timeout;
45+
}
46+
47+
/**
48+
* Overall call timeout in seconds for this vault, including retries.
49+
* <p>
50+
* Takes precedence over the client-wide {@code Skyflow.builder().timeout(...)}. Leave unset
51+
* (null) to inherit that value, or the SDK default of 60s if it is also unset.
52+
*/
53+
public void setTimeout(Integer timeout) {
54+
this.timeout = timeout;
55+
}
56+
57+
public Integer getConnectTimeout() {
58+
return connectTimeout;
59+
}
60+
61+
/**
62+
* Per-attempt connection-establishment timeout in seconds for this vault.
63+
* <p>
64+
* Takes precedence over the client-wide {@code Skyflow.builder().connectTimeout(...)}. Leave
65+
* unset (null) to inherit that value; if neither is set, the underlying HTTP client default
66+
* (10s) applies. Note the overall {@code timeout} still bounds the whole call, including retries.
67+
*/
68+
public void setConnectTimeout(Integer connectTimeout) {
69+
this.connectTimeout = connectTimeout;
70+
}
71+
72+
public Integer getReadTimeout() {
73+
return readTimeout;
74+
}
75+
76+
/**
77+
* Per-attempt response-read timeout in seconds for this vault.
78+
* <p>
79+
* Takes precedence over the client-wide {@code Skyflow.builder().readTimeout(...)}. Leave
80+
* unset (null) to inherit that value; if neither is set, the underlying HTTP client default
81+
* (10s) applies. Note the overall {@code timeout} still bounds the whole call, including retries.
82+
*/
83+
public void setReadTimeout(Integer readTimeout) {
84+
this.readTimeout = readTimeout;
85+
}
86+
87+
public Integer getWriteTimeout() {
88+
return writeTimeout;
89+
}
90+
91+
/**
92+
* Per-attempt request-write timeout in seconds for this vault.
93+
* <p>
94+
* Takes precedence over the client-wide {@code Skyflow.builder().writeTimeout(...)}. Leave
95+
* unset (null) to inherit that value; if neither is set, the underlying HTTP client default
96+
* (10s) applies. Note the overall {@code timeout} still bounds the whole call, including retries.
97+
*/
98+
public void setWriteTimeout(Integer writeTimeout) {
99+
this.writeTimeout = writeTimeout;
100+
}
101+
102+
public Integer getMaxRetries() {
103+
return maxRetries;
104+
}
105+
106+
/**
107+
* Retry attempts after the first failure for this vault.
108+
* <p>
109+
* Takes precedence over the client-wide {@code Skyflow.builder().maxRetries(...)}. Leave unset
110+
* (null) to inherit that value, or the SDK default of 0 if it is also unset — retries are
111+
* opt-in, so non-idempotent bulk writes are not replayed silently.
112+
*/
113+
public void setMaxRetries(Integer maxRetries) {
114+
this.maxRetries = maxRetries;
115+
}
116+
20117
}

0 commit comments

Comments
 (0)