Skip to content

Commit 832c138

Browse files
manusamanusa
committed
fix(vertx): VertxHttpClient uses exclusive Vert.x instance by default
A shared Vert.x instance can still be provided to the VertxHttpClientFactory. This instance will be shared across the different VertxHttpClient instances. It's responsibility of the user to handle the Vert.x shared instance lifecycle Signed-off-by: Marc Nuri <[email protected]>
1 parent ffe667d commit 832c138

8 files changed

Lines changed: 126 additions & 55 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
#### Bugs
66
* Fix #6781: Allowing ipv6 entries to work in NO_PROXY
77
* Fix #6709: VertxHttpClientFactory reuses the same Vertx instance for each VertxHttpClient instance
8+
* Fix #6792: VertxHttpClient uses exclusive Vert.x instance by default
89

910
### 6.13.4 (2024-09-25)
1011

‎httpclient-vertx/src/main/java/io/fabric8/kubernetes/client/vertx/InputStreamReadStream.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ class InputStreamReadStream implements ReadStream<Buffer> {
4141
private Handler<Void> endHandler;
4242
private byte[] bytes;
4343

44-
public InputStreamReadStream(VertxHttpRequest vertxHttpRequest, InputStream is, HttpClientRequest request) {
44+
InputStreamReadStream(VertxHttpRequest vertxHttpRequest, InputStream is, HttpClientRequest request) {
4545
this.vertxHttpRequest = vertxHttpRequest;
4646
this.is = is;
4747
this.request = request;

‎httpclient-vertx/src/main/java/io/fabric8/kubernetes/client/vertx/VertxHttpClient.java‎

Lines changed: 22 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,13 +41,25 @@
4141

4242
public class VertxHttpClient<F extends io.fabric8.kubernetes.client.http.HttpClient.Factory>
4343
extends StandardHttpClient<VertxHttpClient<F>, F, VertxHttpClientBuilder<F>> {
44+
4445
private final Vertx vertx;
4546
private final HttpClient client;
46-
47-
VertxHttpClient(VertxHttpClientBuilder<F> vertxHttpClientBuilder, HttpClient client, AtomicBoolean closed) {
47+
private final boolean closeVertx;
48+
49+
/**
50+
* Create a new VertxHttpClient instance.
51+
*
52+
* @param vertxHttpClientBuilder the builder that created this client.
53+
* @param closed a flag to indicate if the client has been closed.
54+
* @param client the Vert.x HttpClient instance (will be closed alongside the client).
55+
* @param closeVertx whether the Vert.x instance should be closed when the client is closed.
56+
*/
57+
VertxHttpClient(VertxHttpClientBuilder<F> vertxHttpClientBuilder, AtomicBoolean closed, HttpClient client,
58+
boolean closeVertx) {
4859
super(vertxHttpClientBuilder, closed);
4960
this.vertx = vertxHttpClientBuilder.vertx;
5061
this.client = client;
62+
this.closeVertx = closeVertx;
5163
}
5264

5365
HttpClient getClient() {
@@ -69,7 +81,7 @@ public CompletableFuture<WebSocketResponse> buildWebSocketDirect(StandardWebSock
6981
options.setTimeout(request.getTimeout().toMillis());
7082
}
7183

72-
request.headers().entrySet().stream()
84+
request.headers().entrySet()
7385
.forEach(e -> e.getValue().stream().forEach(v -> options.addHeader(e.getKey(), v)));
7486
options.setAbsoluteURI(WebSocket.toWebSocketUri(request.uri()).toString());
7587

@@ -121,7 +133,13 @@ public CompletableFuture<HttpResponse<AsyncBody>> consumeBytesDirect(StandardHtt
121133

122134
@Override
123135
public void doClose() {
124-
client.close();
136+
try {
137+
client.close();
138+
} finally {
139+
if (closeVertx) {
140+
vertx.close();
141+
}
142+
}
125143
}
126144

127145
}

‎httpclient-vertx/src/main/java/io/fabric8/kubernetes/client/vertx/VertxHttpClientBuilder.java‎

Lines changed: 38 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
import io.netty.handler.ssl.IdentityCipherSuiteFilter;
2424
import io.netty.handler.ssl.JdkSslContext;
2525
import io.vertx.core.Vertx;
26+
import io.vertx.core.VertxOptions;
27+
import io.vertx.core.file.FileSystemOptions;
2628
import io.vertx.core.http.HttpVersion;
2729
import io.vertx.core.net.JdkSSLEngineOptions;
2830
import io.vertx.core.net.ProxyOptions;
@@ -37,6 +39,7 @@
3739
import java.util.stream.Stream;
3840

3941
import static io.fabric8.kubernetes.client.utils.HttpClientUtils.decodeBasicCredentials;
42+
import static io.vertx.core.spi.resolver.ResolverProvider.DISABLE_DNS_RESOLVER_PROP_NAME;
4043

4144
public class VertxHttpClientBuilder<F extends HttpClient.Factory>
4245
extends StandardHttpClientBuilder<VertxHttpClient<F>, F, VertxHttpClientBuilder<F>> {
@@ -46,16 +49,25 @@ public class VertxHttpClientBuilder<F extends HttpClient.Factory>
4649
private static final int MAX_WS_MESSAGE_SIZE = Integer.MAX_VALUE;
4750

4851
final Vertx vertx;
52+
private final boolean closeVertx;
4953

50-
public VertxHttpClientBuilder(F clientFactory, Vertx vertx) {
54+
public VertxHttpClientBuilder(F clientFactory, Vertx sharedVertx) {
55+
this(
56+
clientFactory,
57+
sharedVertx != null ? sharedVertx : createVertxInstance(),
58+
sharedVertx == null);
59+
}
60+
61+
VertxHttpClientBuilder(F clientFactory, Vertx vertx, boolean closeVertx) {
5162
super(clientFactory);
5263
this.vertx = vertx;
64+
this.closeVertx = closeVertx;
5365
}
5466

5567
@Override
5668
public VertxHttpClient<F> build() {
5769
if (this.client != null) {
58-
return new VertxHttpClient<>(this, this.client.getClient(), this.client.getClosed());
70+
return new VertxHttpClient<>(this, this.client.getClosed(), this.client.getClient(), closeVertx);
5971
}
6072

6173
WebClientOptions options = new WebClientOptions();
@@ -124,12 +136,12 @@ public SslContextFactory sslContextFactory() {
124136
}
125137
});
126138
}
127-
return new VertxHttpClient<>(this, vertx.createHttpClient(options), new AtomicBoolean());
139+
return new VertxHttpClient<>(this, new AtomicBoolean(), vertx.createHttpClient(options), closeVertx);
128140
}
129141

130142
@Override
131143
protected VertxHttpClientBuilder<F> newInstance(F clientFactory) {
132-
return new VertxHttpClientBuilder<>(clientFactory, vertx);
144+
return new VertxHttpClientBuilder<>(clientFactory, vertx, closeVertx);
133145
}
134146

135147
private ProxyType convertProxyType() {
@@ -145,4 +157,26 @@ private ProxyType convertProxyType() {
145157
}
146158
}
147159

160+
private static Vertx createVertxInstance() {
161+
// We must disable the async DNS resolver as it can cause issues when resolving the Vault instance.
162+
// This is done using the DISABLE_DNS_RESOLVER_PROP_NAME system property.
163+
// The DNS resolver used by vert.x is configured during the (synchronous) initialization.
164+
// So, we just need to disable the async resolver around the Vert.x instance creation.
165+
final String originalValue = System.getProperty(DISABLE_DNS_RESOLVER_PROP_NAME);
166+
Vertx vertx;
167+
try {
168+
System.setProperty(DISABLE_DNS_RESOLVER_PROP_NAME, "true");
169+
vertx = Vertx.vertx(new VertxOptions()
170+
.setFileSystemOptions(new FileSystemOptions().setFileCachingEnabled(false).setClassPathResolvingEnabled(false))
171+
.setUseDaemonThread(true));
172+
} finally {
173+
// Restore the original value
174+
if (originalValue == null) {
175+
System.clearProperty(DISABLE_DNS_RESOLVER_PROP_NAME);
176+
} else {
177+
System.setProperty(DISABLE_DNS_RESOLVER_PROP_NAME, originalValue);
178+
}
179+
}
180+
return vertx;
181+
}
148182
}

‎httpclient-vertx/src/main/java/io/fabric8/kubernetes/client/vertx/VertxHttpClientFactory.java‎

Lines changed: 13 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -17,60 +17,31 @@
1717

1818
import io.fabric8.kubernetes.client.Config;
1919
import io.vertx.core.Vertx;
20-
import io.vertx.core.VertxOptions;
21-
import io.vertx.core.file.FileSystemOptions;
2220
import io.vertx.ext.web.client.WebClientOptions;
2321

24-
import static io.vertx.core.spi.resolver.ResolverProvider.DISABLE_DNS_RESOLVER_PROP_NAME;
25-
2622
public class VertxHttpClientFactory implements io.fabric8.kubernetes.client.http.HttpClient.Factory {
2723

28-
private static final class VertxHolder {
29-
30-
private static final Vertx INSTANCE = createVertxInstance();
31-
32-
private static synchronized Vertx createVertxInstance() {
33-
// We must disable the async DNS resolver as it can cause issues when resolving the Vault instance.
34-
// This is done using the DISABLE_DNS_RESOLVER_PROP_NAME system property.
35-
// The DNS resolver used by vert.x is configured during the (synchronous) initialization.
36-
// So, we just need to disable the async resolver around the Vert.x instance creation.
37-
final String originalValue = System.getProperty(DISABLE_DNS_RESOLVER_PROP_NAME);
38-
Vertx vertx;
39-
try {
40-
System.setProperty(DISABLE_DNS_RESOLVER_PROP_NAME, "true");
41-
vertx = Vertx.vertx(new VertxOptions()
42-
.setFileSystemOptions(new FileSystemOptions().setFileCachingEnabled(false).setClassPathResolvingEnabled(false))
43-
.setUseDaemonThread(true));
44-
} finally {
45-
// Restore the original value
46-
if (originalValue == null) {
47-
System.clearProperty(DISABLE_DNS_RESOLVER_PROP_NAME);
48-
} else {
49-
System.setProperty(DISABLE_DNS_RESOLVER_PROP_NAME, originalValue);
50-
}
51-
}
52-
return vertx;
53-
}
54-
}
55-
56-
private final Vertx vertx;
24+
final Vertx sharedVertx;
5725

5826
public VertxHttpClientFactory() {
59-
this(VertxHolder.INSTANCE);
60-
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
61-
if (vertx != null) {
62-
vertx.close();
63-
}
64-
}));
27+
this(null);
6528
}
6629

67-
public VertxHttpClientFactory(Vertx vertx) {
68-
this.vertx = vertx;
30+
/**
31+
* Create a new instance of the factory that will reuse the provided {@link Vertx} instance.
32+
* <p>
33+
* It's the user's responsibility to manage the lifecycle of the provided Vert.x instance.
34+
* Operations such as close, and so on are left on hands of the user.
35+
*
36+
* @param sharedVertx the Vertx instance to use.
37+
*/
38+
public VertxHttpClientFactory(Vertx sharedVertx) {
39+
this.sharedVertx = sharedVertx;
6940
}
7041

7142
@Override
7243
public VertxHttpClientBuilder<VertxHttpClientFactory> newBuilder() {
73-
return new VertxHttpClientBuilder<>(this, vertx);
44+
return new VertxHttpClientBuilder<>(this, sharedVertx);
7445
}
7546

7647
/**

‎httpclient-vertx/src/main/java/io/fabric8/kubernetes/client/vertx/VertxHttpRequest.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232

3333
import java.io.InputStream;
3434
import java.nio.ByteBuffer;
35-
import java.util.Arrays;
3635
import java.util.LinkedHashMap;
3736
import java.util.List;
3837
import java.util.Map;
@@ -77,7 +76,7 @@ public Optional<HttpResponse<?>> previousResponse() {
7776

7877
final Vertx vertx;
7978
private final RequestOptions options;
80-
private StandardHttpRequest request;
79+
private final StandardHttpRequest request;
8180

8281
public VertxHttpRequest(Vertx vertx, RequestOptions options, StandardHttpRequest request) {
8382
this.vertx = vertx;
@@ -113,7 +112,7 @@ public void cancel() {
113112
};
114113
resp.handler(buffer -> {
115114
try {
116-
consumer.consume(Arrays.asList(ByteBuffer.wrap(buffer.getBytes())), result);
115+
consumer.consume(List.of(ByteBuffer.wrap(buffer.getBytes())), result);
117116
} catch (Exception e) {
118117
resp.request().reset();
119118
result.done().completeExceptionally(e);

‎httpclient-vertx/src/test/java/io/fabric8/kubernetes/client/vertx/VertxHttpClientBuilderTest.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,14 @@
1616
package io.fabric8.kubernetes.client.vertx;
1717

1818
import io.fabric8.kubernetes.client.http.HttpClient;
19+
import io.vertx.core.Vertx;
20+
import io.vertx.core.impl.VertxImpl;
21+
import org.assertj.core.api.InstanceOfAssertFactories;
1922
import org.junit.jupiter.api.Test;
2023

2124
import java.util.concurrent.TimeUnit;
2225

26+
import static org.assertj.core.api.Assertions.assertThat;
2327
import static org.junit.jupiter.api.Assertions.assertNotNull;
2428

2529
class VertxHttpClientBuilderTest {
@@ -35,4 +39,46 @@ void testZeroTimeouts() {
3539
}
3640
}
3741

42+
@Test
43+
void reusesVertxInstanceWhenSharedVertx() {
44+
Vertx vertx = Vertx.vertx();
45+
try (HttpClient client = new VertxHttpClientFactory(vertx).newBuilder().build()) {
46+
assertThat(client)
47+
.isInstanceOf(VertxHttpClient.class)
48+
.extracting("vertx")
49+
.isSameAs(vertx);
50+
} finally {
51+
vertx.close();
52+
}
53+
}
54+
55+
@Test
56+
void createsVertxInstanceWhenNoSharedVertx() {
57+
try (HttpClient client = new VertxHttpClientFactory().newBuilder().build()) {
58+
assertThat(client)
59+
.isInstanceOf(VertxHttpClient.class)
60+
.extracting("vertx")
61+
.isNotNull();
62+
}
63+
}
64+
65+
@Test
66+
void doesntCloseSharedVertxInstanceWhenClientIsClosed() {
67+
final Vertx vertx = Vertx.vertx();
68+
final var builder = new VertxHttpClientFactory(vertx).newBuilder();
69+
builder.build().close();
70+
assertThat(builder.vertx)
71+
.asInstanceOf(InstanceOfAssertFactories.type(VertxImpl.class))
72+
.returns(false, vi -> vi.closeFuture().isClosed());
73+
vertx.close();
74+
}
75+
76+
@Test
77+
void closesVertxInstanceWhenClientIsClosed() {
78+
final var builder = new VertxHttpClientFactory().newBuilder();
79+
builder.build().close();
80+
assertThat(builder.vertx)
81+
.asInstanceOf(InstanceOfAssertFactories.type(VertxImpl.class))
82+
.returns(true, vi -> vi.closeFuture().isClosed());
83+
}
3884
}

‎kubernetes-tests/src/test/java/io/fabric8/kubernetes/client/mock/DefaultSharedIndexInformerTest.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1205,7 +1205,7 @@ void testCustomExceptionHandler() throws InterruptedException {
12051205
}
12061206

12071207
@Test
1208-
void testClientStopClosesInformer() throws InterruptedException {
1208+
void testClientStopClosesInformer() throws Exception {
12091209
// Given
12101210
setupMockServerExpectations(Animal.class, "ns1", this::getList,
12111211
r -> new WatchEvent(getAnimal("red-panda", "Carnivora", r), "ADDED"), null, null);
@@ -1218,6 +1218,8 @@ void testClientStopClosesInformer() throws InterruptedException {
12181218

12191219
animalSharedIndexInformer.start();
12201220

1221+
await().atMost(10, TimeUnit.SECONDS).until(animalSharedIndexInformer::hasSynced);
1222+
12211223
client.close();
12221224

12231225
await().atMost(60, TimeUnit.SECONDS).until(() -> animalSharedIndexInformer.stopped().toCompletableFuture().isDone());

0 commit comments

Comments
 (0)