Skip to content

Commit d186946

Browse files
3pacccccc3pacccccc
authored
[feat][client] PIP-234: Support shared resources in PulsarAdmin to reduce thread usage (#24893)
1 parent e557b00 commit d186946

File tree

7 files changed

+228
-23
lines changed

7 files changed

+228
-23
lines changed

‎pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/PulsarAdminBuilder.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.pulsar.client.api.Authentication;
2525
import org.apache.pulsar.client.api.PulsarClientException;
2626
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
27+
import org.apache.pulsar.client.api.PulsarClientSharedResources;
2728

2829
/**
2930
* Builder class for a {@link PulsarAdmin} instance.
@@ -393,4 +394,17 @@ PulsarAdminBuilder authentication(String authPluginClassName, Map<String, String
393394
* @throws IllegalArgumentException if the length of description exceeds 64
394395
*/
395396
PulsarAdminBuilder description(String description);
397+
398+
/**
399+
* Provide a set of shared client resources to be reused by this client.
400+
* <p>
401+
* Providing a shared resource instance allows PulsarClient instances to share resources
402+
* (only support IO/event loops, timers, DNS resolver/cache) with other PulsarClient
403+
* instances, reducing memory footprint and thread usage when creating many clients in the same JVM.
404+
*
405+
* @param sharedResources the shared resources instance created with {@link PulsarClientSharedResources#builder()}
406+
* @return the adminClient builder instance
407+
*/
408+
PulsarAdminBuilder sharedResources(PulsarClientSharedResources sharedResources);
409+
396410
}

‎pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImpl.java‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@
2929
import org.apache.pulsar.client.api.AuthenticationFactory;
3030
import org.apache.pulsar.client.api.PulsarClientException;
3131
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
32+
import org.apache.pulsar.client.api.PulsarClientSharedResources;
33+
import org.apache.pulsar.client.impl.PulsarClientSharedResourcesImpl;
3234
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
3335
import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
3436

@@ -39,10 +41,12 @@ public class PulsarAdminBuilderImpl implements PulsarAdminBuilder {
3941

4042
private ClassLoader clientBuilderClassLoader = null;
4143
private boolean acceptGzipCompression = true;
44+
private transient PulsarClientSharedResourcesImpl sharedResources;
4245

4346
@Override
4447
public PulsarAdmin build() throws PulsarClientException {
45-
return new PulsarAdminImpl(conf.getServiceUrl(), conf, clientBuilderClassLoader, acceptGzipCompression);
48+
return new PulsarAdminImpl(conf.getServiceUrl(), conf,
49+
clientBuilderClassLoader, acceptGzipCompression, sharedResources);
4650
}
4751

4852
public PulsarAdminBuilderImpl() {
@@ -292,4 +296,10 @@ public PulsarAdminBuilder description(String description) {
292296
this.conf.setDescription(description);
293297
return this;
294298
}
299+
300+
@Override
301+
public PulsarAdminBuilder sharedResources(PulsarClientSharedResources sharedResources) {
302+
this.sharedResources = (PulsarClientSharedResourcesImpl) sharedResources;
303+
return this;
304+
}
295305
}

‎pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import javax.ws.rs.client.Client;
2727
import javax.ws.rs.client.ClientBuilder;
2828
import javax.ws.rs.client.WebTarget;
29+
import lombok.Getter;
2930
import org.apache.commons.lang3.StringUtils;
3031
import org.apache.pulsar.client.admin.Bookies;
3132
import org.apache.pulsar.client.admin.BrokerStats;
@@ -56,6 +57,7 @@
5657
import org.apache.pulsar.client.api.Authentication;
5758
import org.apache.pulsar.client.api.AuthenticationFactory;
5859
import org.apache.pulsar.client.api.PulsarClientException;
60+
import org.apache.pulsar.client.impl.PulsarClientSharedResourcesImpl;
5961
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
6062
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
6163
import org.apache.pulsar.common.net.ServiceURI;
@@ -91,6 +93,7 @@ public class PulsarAdminImpl implements PulsarAdmin {
9193
private final ResourceQuotas resourceQuotas;
9294
private final ClientConfigurationData clientConfigData;
9395
private final Client client;
96+
@Getter
9497
private final AsyncHttpConnector asyncHttpConnector;
9598
private final String serviceUrl;
9699
private final Lookup lookups;
@@ -106,11 +109,12 @@ public class PulsarAdminImpl implements PulsarAdmin {
106109

107110
public PulsarAdminImpl(String serviceUrl, ClientConfigurationData clientConfigData,
108111
ClassLoader clientBuilderClassLoader) throws PulsarClientException {
109-
this(serviceUrl, clientConfigData, clientBuilderClassLoader, true);
112+
this(serviceUrl, clientConfigData, clientBuilderClassLoader, true, null);
110113
}
111114

112115
public PulsarAdminImpl(String serviceUrl, ClientConfigurationData clientConfigData,
113-
ClassLoader clientBuilderClassLoader, boolean acceptGzipCompression)
116+
ClassLoader clientBuilderClassLoader, boolean acceptGzipCompression,
117+
PulsarClientSharedResourcesImpl sharedResources)
114118
throws PulsarClientException {
115119
checkArgument(StringUtils.isNotBlank(serviceUrl), "Service URL needs to be specified");
116120

@@ -157,7 +161,7 @@ public PulsarAdminImpl(String serviceUrl, ClientConfigurationData clientConfigDa
157161
Math.toIntExact(clientConfigData.getConnectionTimeoutMs()),
158162
Math.toIntExact(clientConfigData.getReadTimeoutMs()),
159163
Math.toIntExact(clientConfigData.getRequestTimeoutMs()),
160-
clientConfigData.getAutoCertRefreshSeconds());
164+
clientConfigData.getAutoCertRefreshSeconds(), sharedResources);
161165

162166
long requestTimeoutMs = clientConfigData.getRequestTimeoutMs();
163167
this.clusters = new ClustersImpl(root, auth, requestTimeoutMs);

‎pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java‎

Lines changed: 92 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -28,12 +28,15 @@
2828
import static org.asynchttpclient.util.HttpConstants.ResponseStatusCodes.TEMPORARY_REDIRECT_307;
2929
import static org.asynchttpclient.util.MiscUtils.isNonEmpty;
3030
import com.spotify.futures.ConcurrencyReducer;
31+
import io.netty.channel.EventLoopGroup;
3132
import io.netty.handler.codec.http.DefaultHttpHeaders;
3233
import io.netty.handler.codec.http.HttpRequest;
3334
import io.netty.handler.codec.http.HttpResponse;
35+
import io.netty.resolver.NameResolver;
3436
import io.netty.util.concurrent.DefaultThreadFactory;
3537
import java.io.ByteArrayOutputStream;
3638
import java.io.IOException;
39+
import java.net.InetAddress;
3740
import java.net.InetSocketAddress;
3841
import java.net.URI;
3942
import java.security.GeneralSecurityException;
@@ -53,20 +56,25 @@
5356
import javax.ws.rs.client.Client;
5457
import javax.ws.rs.core.HttpHeaders;
5558
import javax.ws.rs.core.Response.Status;
59+
import lombok.Data;
5660
import lombok.Getter;
5761
import lombok.SneakyThrows;
5862
import lombok.extern.slf4j.Slf4j;
5963
import org.apache.commons.lang3.Validate;
6064
import org.apache.pulsar.PulsarVersion;
6165
import org.apache.pulsar.client.admin.internal.PulsarAdminImpl;
6266
import org.apache.pulsar.client.api.PulsarClientException;
67+
import org.apache.pulsar.client.impl.PulsarClientSharedResourcesImpl;
6368
import org.apache.pulsar.client.impl.PulsarServiceNameResolver;
6469
import org.apache.pulsar.client.impl.ServiceNameResolver;
6570
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
71+
import org.apache.pulsar.client.util.ExecutorProvider;
6672
import org.apache.pulsar.client.util.PulsarHttpAsyncSslEngineFactory;
6773
import org.apache.pulsar.common.util.FutureUtil;
6874
import org.apache.pulsar.common.util.PulsarSslConfiguration;
6975
import org.apache.pulsar.common.util.PulsarSslFactory;
76+
import org.apache.pulsar.common.util.netty.DnsResolverUtil;
77+
import org.apache.pulsar.common.util.netty.EventLoopUtil;
7078
import org.asynchttpclient.AsyncCompletionHandlerBase;
7179
import org.asynchttpclient.AsyncHandler;
7280
import org.asynchttpclient.AsyncHttpClient;
@@ -103,6 +111,10 @@ public class AsyncHttpConnector implements Connector, AsyncHttpRequestExecutor {
103111
new DefaultThreadFactory("delayer"));
104112
private ScheduledExecutorService sslRefresher;
105113
private final boolean acceptGzipCompression;
114+
@Getter
115+
private final NameResolver<InetAddress> nameResolver;
116+
private final EventLoopGroup eventLoopGroup;
117+
private final boolean createdEventLoopGroup;
106118
private final Map<String, ConcurrencyReducer<Response>> concurrencyReducers = new ConcurrentHashMap<>();
107119
private PulsarSslFactory sslFactory;
108120

@@ -112,33 +124,66 @@ public AsyncHttpConnector(Client client, ClientConfigurationData conf, int autoC
112124
(int) client.getConfiguration().getProperty(ClientProperties.READ_TIMEOUT),
113125
PulsarAdminImpl.DEFAULT_REQUEST_TIMEOUT_SECONDS * 1000,
114126
autoCertRefreshTimeSeconds,
115-
conf, acceptGzipCompression);
127+
conf, acceptGzipCompression, null);
116128
}
117129

118130
@SneakyThrows
119131
public AsyncHttpConnector(int connectTimeoutMs, int readTimeoutMs,
120132
int requestTimeoutMs,
121133
int autoCertRefreshTimeSeconds, ClientConfigurationData conf,
122-
boolean acceptGzipCompression) {
134+
boolean acceptGzipCompression,
135+
PulsarClientSharedResourcesImpl sharedResources) {
123136
Validate.notEmpty(conf.getServiceUrl(), "Service URL is not provided");
124137
serviceNameResolver = new PulsarServiceNameResolver();
125138
String serviceUrl = conf.getServiceUrl();
126139
serviceNameResolver.updateServiceUrl(serviceUrl);
127140
this.acceptGzipCompression = acceptGzipCompression;
141+
SharedResourceHolder sharedResourceHolder =
142+
buildResourcesIfConfigured(sharedResources);
143+
this.nameResolver = sharedResourceHolder.getNameResolver();
144+
this.eventLoopGroup = sharedResourceHolder.getEventLoopGroup();
145+
this.createdEventLoopGroup = sharedResourceHolder.isCreateEventLoop();
128146
AsyncHttpClientConfig asyncHttpClientConfig =
129147
createAsyncHttpClientConfig(conf, connectTimeoutMs, readTimeoutMs, requestTimeoutMs,
130-
autoCertRefreshTimeSeconds);
148+
autoCertRefreshTimeSeconds, sharedResources);
131149
httpClient = createAsyncHttpClient(asyncHttpClientConfig);
132150
this.requestTimeout = requestTimeoutMs > 0 ? Duration.ofMillis(requestTimeoutMs) : null;
133151
this.maxRetries = httpClient.getConfig().getMaxRequestRetry();
134152
}
135153

154+
private SharedResourceHolder buildResourcesIfConfigured(
155+
PulsarClientSharedResourcesImpl sharedResources) {
156+
EventLoopGroup eventLoopGroup = null;
157+
NameResolver<InetAddress> nameResolver = null;
158+
boolean createdEventLoopGroup = false;
159+
if (sharedResources != null && sharedResources.getDnsResolverGroup() != null) {
160+
if (sharedResources.getIoEventLoopGroup() != null) {
161+
eventLoopGroup = sharedResources.getIoEventLoopGroup();
162+
} else {
163+
// build an EventLoopGroup with default value
164+
eventLoopGroup = EventLoopUtil.newEventLoopGroup(
165+
Runtime.getRuntime().availableProcessors(), false,
166+
new ExecutorProvider.ExtendedThreadFactory("pulsar-admin-client-io",
167+
Thread.currentThread().isDaemon()));
168+
createdEventLoopGroup = true;
169+
}
170+
nameResolver = DnsResolverUtil.adaptToNameResolver(
171+
sharedResources.getDnsResolverGroup().createAddressResolver(eventLoopGroup));
172+
} else {
173+
return SharedResourceHolder.EMPTY;
174+
}
175+
return new SharedResourceHolder(nameResolver, eventLoopGroup, createdEventLoopGroup);
176+
}
177+
136178
private AsyncHttpClientConfig createAsyncHttpClientConfig(ClientConfigurationData conf, int connectTimeoutMs,
137179
int readTimeoutMs,
138-
int requestTimeoutMs, int autoCertRefreshTimeSeconds)
180+
int requestTimeoutMs,
181+
int autoCertRefreshTimeSeconds,
182+
PulsarClientSharedResourcesImpl sharedResources)
139183
throws GeneralSecurityException, IOException {
140184
DefaultAsyncHttpClientConfig.Builder confBuilder = new DefaultAsyncHttpClientConfig.Builder();
141-
configureAsyncHttpClientConfig(conf, connectTimeoutMs, readTimeoutMs, requestTimeoutMs, confBuilder);
185+
configureAsyncHttpClientConfig(conf, connectTimeoutMs,
186+
readTimeoutMs, requestTimeoutMs, confBuilder, sharedResources);
142187
if (conf.getServiceUrl().startsWith("https://")) {
143188
configureAsyncHttpClientSslEngineFactory(conf, autoCertRefreshTimeSeconds, confBuilder);
144189
}
@@ -148,7 +193,8 @@ private AsyncHttpClientConfig createAsyncHttpClientConfig(ClientConfigurationDat
148193

149194
private void configureAsyncHttpClientConfig(ClientConfigurationData conf, int connectTimeoutMs, int readTimeoutMs,
150195
int requestTimeoutMs,
151-
DefaultAsyncHttpClientConfig.Builder confBuilder) {
196+
DefaultAsyncHttpClientConfig.Builder confBuilder,
197+
PulsarClientSharedResourcesImpl sharedResources) {
152198
if (conf.getConnectionsPerBroker() > 0) {
153199
confBuilder.setMaxConnectionsPerHost(conf.getConnectionsPerBroker());
154200
// Use the request timeout value for acquireFreeChannelTimeout so that we don't need to add
@@ -159,6 +205,14 @@ private void configureAsyncHttpClientConfig(ClientConfigurationData conf, int co
159205
if (conf.getConnectionMaxIdleSeconds() > 0) {
160206
confBuilder.setPooledConnectionIdleTimeout(conf.getConnectionMaxIdleSeconds() * 1000);
161207
}
208+
if (sharedResources != null) {
209+
if (this.eventLoopGroup != null) {
210+
confBuilder.setEventLoopGroup(this.eventLoopGroup);
211+
}
212+
if (sharedResources.getTimer() != null) {
213+
confBuilder.setNettyTimer(sharedResources.getTimer());
214+
}
215+
}
162216
confBuilder.setCookieStore(null);
163217
confBuilder.setUseProxyProperties(true);
164218
confBuilder.setFollowRedirect(false);
@@ -177,7 +231,7 @@ public boolean keepAlive(InetSocketAddress remoteAddress, Request ahcRequest,
177231
HttpRequest request, HttpResponse response) {
178232
// Close connection upon a server error or per HTTP spec
179233
return (response.status().code() / 100 != 5)
180-
&& super.keepAlive(remoteAddress, ahcRequest, request, response);
234+
&& super.keepAlive(remoteAddress, ahcRequest, request, response);
181235
}
182236
});
183237
confBuilder.setDisableHttpsEndpointIdentificationAlgorithm(!conf.isTlsHostnameVerificationEnable());
@@ -331,9 +385,9 @@ private <T> void retryOperation(
331385
throwable);
332386
}
333387
resultFuture.completeExceptionally(
334-
new RetryException("Could not complete the operation. Number of retries "
335-
+ "has been exhausted. Failed reason: " + throwable.getMessage(),
336-
throwable));
388+
new RetryException("Could not complete the operation. Number of retries "
389+
+ "has been exhausted. Failed reason: " + throwable.getMessage(),
390+
throwable));
337391
}
338392
}
339393
} else {
@@ -376,7 +430,7 @@ public CompletableFuture<Response> executeRequest(Request request) {
376430
}
377431

378432
public CompletableFuture<Response> executeRequest(Request request,
379-
Supplier<AsyncHandler<Response>> handlerSupplier) {
433+
Supplier<AsyncHandler<Response>> handlerSupplier) {
380434
return executeRequest(request, handlerSupplier, 0);
381435
}
382436

@@ -426,14 +480,17 @@ private CompletableFuture<Response> executeRedirect(Request request, Response re
426480
if (switchToGet) {
427481
builder.setMethod(GET);
428482
}
483+
if (this.nameResolver != null) {
484+
builder.setNameResolver(this.nameResolver);
485+
}
429486
builder.setUri(newUri);
430487
if (keepBody) {
431488
builder.setCharset(request.getCharset());
432489
if (isNonEmpty(request.getFormParams())) {
433490
builder.setFormParams(request.getFormParams());
434491
} else if (request.getStringData() != null) {
435492
builder.setBody(request.getStringData());
436-
} else if (request.getByteData() != null){
493+
} else if (request.getByteData() != null) {
437494
builder.setBody(request.getByteData());
438495
} else if (request.getByteBufferData() != null) {
439496
builder.setBody(request.getByteBufferData());
@@ -485,6 +542,9 @@ private Request prepareRequest(InetSocketAddress host, ClientRequest request) th
485542
BoundRequestBuilder builder =
486543
httpClient.prepare(currentRequest.getMethod(), currentRequest.getUri().toString());
487544

545+
if (this.nameResolver != null) {
546+
builder.setNameResolver(this.nameResolver);
547+
}
488548
if (currentRequest.hasEntity()) {
489549
ByteArrayOutputStream outStream = new ByteArrayOutputStream();
490550
currentRequest.setStreamProvider(contentLength -> outStream);
@@ -518,6 +578,9 @@ public void close() {
518578
if (sslRefresher != null) {
519579
sslRefresher.shutdownNow();
520580
}
581+
if (createdEventLoopGroup && eventLoopGroup != null && !eventLoopGroup.isShutdown()) {
582+
eventLoopGroup.shutdownGracefully();
583+
}
521584
} catch (IOException e) {
522585
log.warn("Failed to close http client", e);
523586
}
@@ -556,4 +619,21 @@ protected void refreshSslContext() {
556619
}
557620
}
558621

622+
@Data
623+
private static class SharedResourceHolder {
624+
static final SharedResourceHolder EMPTY = new SharedResourceHolder(null, null, false);
625+
626+
final NameResolver<InetAddress> nameResolver;
627+
final EventLoopGroup eventLoopGroup;
628+
final boolean createEventLoop;
629+
630+
SharedResourceHolder(NameResolver<InetAddress> nameResolver,
631+
EventLoopGroup eventLoopGroup,
632+
boolean createEventLoop) {
633+
this.nameResolver = nameResolver;
634+
this.eventLoopGroup = eventLoopGroup;
635+
this.createEventLoop = createEventLoop;
636+
}
637+
}
638+
559639
}

‎pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorProvider.java‎

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

2121
import javax.ws.rs.client.Client;
2222
import javax.ws.rs.core.Configuration;
23+
import org.apache.pulsar.client.impl.PulsarClientSharedResourcesImpl;
2324
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
2425
import org.glassfish.jersey.client.spi.Connector;
2526
import org.glassfish.jersey.client.spi.ConnectorProvider;
@@ -51,8 +52,8 @@ public Connector getConnector(Client client, Configuration runtimeConfig) {
5152

5253

5354
public AsyncHttpConnector getConnector(int connectTimeoutMs, int readTimeoutMs, int requestTimeoutMs,
54-
int autoCertRefreshTimeSeconds) {
55+
int autoCertRefreshTimeSeconds, PulsarClientSharedResourcesImpl sharedResources) {
5556
return new AsyncHttpConnector(connectTimeoutMs, readTimeoutMs, requestTimeoutMs, autoCertRefreshTimeSeconds,
56-
conf, acceptGzipCompression);
57+
conf, acceptGzipCompression, sharedResources);
5758
}
5859
}

0 commit comments

Comments
 (0)