You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "3pacccccc (via GitHub)" <gi...@apache.org> on 2025/10/25 16:15:05 UTC
[PR] [feat][client]Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
3pacccccc opened a new pull request, #24893:
URL: https://github.com/apache/pulsar/pull/24893
Main Issue: https://github.com/apache/pulsar/issues/24796
### Motivation
When creating multiple PulsarAdmin clients in the same JVM, each client currently creates its own set of resources (event loops, timers, DNS resolvers). This leads to excessive thread consumption and memory usage when many admin clients are instantiated.
This change allows PulsarAdmin clients to share resources through the `PulsarClientSharedResources` mechanism, similar to what's already available for PulsarClient.
### Modifications
1. Added `sharedResources(PulsarClientSharedResources)` method to `PulsarAdminBuilder` API
2. Modified `PulsarAdminBuilderImpl` to accept and propagate shared resources configuration
3. Updated `PulsarAdminImpl` constructor to pass shared resources to HTTP connector
4. Enhanced `AsyncHttpConnector` to utilize shared resources:
- Event loop group for I/O operations
- Netty timer for timeouts and scheduling
- DNS resolver for name resolution
5. Added test to verify shared resources are properly reused across multiple admin clients.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [x] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: https://github.com/3pacccccc/pulsar/pull/29
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client]Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24893:
URL: https://github.com/apache/pulsar/pull/24893#discussion_r2467738048
##########
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java:
##########
@@ -112,33 +120,57 @@ public AsyncHttpConnector(Client client, ClientConfigurationData conf, int autoC
(int) client.getConfiguration().getProperty(ClientProperties.READ_TIMEOUT),
PulsarAdminImpl.DEFAULT_REQUEST_TIMEOUT_SECONDS * 1000,
autoCertRefreshTimeSeconds,
- conf, acceptGzipCompression);
+ conf, acceptGzipCompression, null);
}
@SneakyThrows
public AsyncHttpConnector(int connectTimeoutMs, int readTimeoutMs,
int requestTimeoutMs,
int autoCertRefreshTimeSeconds, ClientConfigurationData conf,
- boolean acceptGzipCompression) {
+ boolean acceptGzipCompression,
+ PulsarClientSharedResourcesImpl sharedResources) {
Validate.notEmpty(conf.getServiceUrl(), "Service URL is not provided");
serviceNameResolver = new PulsarServiceNameResolver();
String serviceUrl = conf.getServiceUrl();
serviceNameResolver.updateServiceUrl(serviceUrl);
this.acceptGzipCompression = acceptGzipCompression;
AsyncHttpClientConfig asyncHttpClientConfig =
createAsyncHttpClientConfig(conf, connectTimeoutMs, readTimeoutMs, requestTimeoutMs,
- autoCertRefreshTimeSeconds);
+ autoCertRefreshTimeSeconds, sharedResources);
httpClient = createAsyncHttpClient(asyncHttpClientConfig);
this.requestTimeout = requestTimeoutMs > 0 ? Duration.ofMillis(requestTimeoutMs) : null;
this.maxRetries = httpClient.getConfig().getMaxRequestRetry();
+ this.nameResolver = buildNameResolverIfConfigured(sharedResources);
+ }
+ private NameResolver<InetAddress> buildNameResolverIfConfigured(PulsarClientSharedResourcesImpl sharedResources) {
+ if (sharedResources != null && sharedResources.getDnsResolverGroup() != null) {
+ EventLoopGroup eventLoopGroupReference;
+ if (sharedResources.getIoEventLoopGroup() != null) {
+ eventLoopGroupReference = sharedResources.getIoEventLoopGroup();
+ } else {
+ // build an EventLoopGroup with default value
+ eventLoopGroupReference = EventLoopUtil.newEventLoopGroup(
+ Runtime.getRuntime().availableProcessors(), false,
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-admin-client-io",
+ Thread.currentThread().isDaemon()));
Review Comment:
OK, good catch, I'll fix it. @lhotari
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-234: Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #24893:
URL: https://github.com/apache/pulsar/pull/24893
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-234: Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #24893:
URL: https://github.com/apache/pulsar/pull/24893#discussion_r2480398530
##########
pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java:
##########
@@ -188,6 +195,96 @@ public void testClientDescription() throws PulsarClientException {
PulsarAdmin.builder().serviceHttpUrl("http://localhost:8080").description("forked").build();
}
+ @Test
+ public void testClientBuildWithSharedResources() throws PulsarClientException {
+ PulsarClientSharedResources sharedResources = PulsarClientSharedResources.builder()
+ .configureEventLoop(eventLoopGroupConfig -> {
+ eventLoopGroupConfig
+ .name("testEventLoop")
+ .numberOfThreads(20);
+ })
+ .configureDnsResolver(dnsResolverConfig -> {
+ dnsResolverConfig.localAddress(new InetSocketAddress(0));
+ })
+ .configureTimer(timerConfig -> {
+ timerConfig.name("testTimer").tickDuration(100, TimeUnit.MILLISECONDS);
+ })
+ .build();
+ // create two adminClients and check if they share the same event loop group and netty timer
+ @Cleanup
+ PulsarAdminImpl pulsarAdminImpl1 =
+ (PulsarAdminImpl) PulsarAdmin.builder()
+ .serviceHttpUrl("http://localhost:8080")
+ .sharedResources(sharedResources)
+ .build();
+ @Cleanup
+ PulsarAdminImpl pulsarAdminImpl2 =
+ (PulsarAdminImpl) PulsarAdmin.builder()
+ .serviceHttpUrl("http://localhost:8080")
+ .sharedResources(sharedResources)
+ .build();
+
+ EventLoopGroup eventLoopGroup1 =
+ pulsarAdminImpl1.getAsyncHttpConnector().getHttpClient().getConfig().getEventLoopGroup();
+ EventLoopGroup eventLoopGroup2 =
+ pulsarAdminImpl2.getAsyncHttpConnector().getHttpClient().getConfig().getEventLoopGroup();
+ Timer nettyTimer1 = pulsarAdminImpl1.getAsyncHttpConnector().getHttpClient().getConfig().getNettyTimer();
+ Timer nettyTimer2 = pulsarAdminImpl2.getAsyncHttpConnector().getHttpClient().getConfig().getNettyTimer();
+ boolean b1 = eventLoopGroup1 == eventLoopGroup2;
+ boolean b2 = nettyTimer1 == nettyTimer2;
+ assertThat(b1 && b2).isTrue();
Review Comment:
Intermediate boolean variables `b1` and `b2` are unnecessary. Simplify to `assertThat(eventLoopGroup1).isSameAs(eventLoopGroup2)` and `assertThat(nettyTimer1).isSameAs(nettyTimer2)` for better readability and clearer assertion messages on failure.
```suggestion
assertThat(eventLoopGroup1).isSameAs(eventLoopGroup2);
assertThat(nettyTimer1).isSameAs(nettyTimer2);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-234: Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24893:
URL: https://github.com/apache/pulsar/pull/24893#discussion_r2480568763
##########
pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java:
##########
@@ -188,6 +195,96 @@ public void testClientDescription() throws PulsarClientException {
PulsarAdmin.builder().serviceHttpUrl("http://localhost:8080").description("forked").build();
}
+ @Test
+ public void testClientBuildWithSharedResources() throws PulsarClientException {
+ PulsarClientSharedResources sharedResources = PulsarClientSharedResources.builder()
+ .configureEventLoop(eventLoopGroupConfig -> {
+ eventLoopGroupConfig
+ .name("testEventLoop")
+ .numberOfThreads(20);
+ })
+ .configureDnsResolver(dnsResolverConfig -> {
+ dnsResolverConfig.localAddress(new InetSocketAddress(0));
+ })
+ .configureTimer(timerConfig -> {
+ timerConfig.name("testTimer").tickDuration(100, TimeUnit.MILLISECONDS);
+ })
+ .build();
+ // create two adminClients and check if they share the same event loop group and netty timer
+ @Cleanup
+ PulsarAdminImpl pulsarAdminImpl1 =
+ (PulsarAdminImpl) PulsarAdmin.builder()
+ .serviceHttpUrl("http://localhost:8080")
+ .sharedResources(sharedResources)
+ .build();
+ @Cleanup
+ PulsarAdminImpl pulsarAdminImpl2 =
+ (PulsarAdminImpl) PulsarAdmin.builder()
+ .serviceHttpUrl("http://localhost:8080")
+ .sharedResources(sharedResources)
+ .build();
+
+ EventLoopGroup eventLoopGroup1 =
+ pulsarAdminImpl1.getAsyncHttpConnector().getHttpClient().getConfig().getEventLoopGroup();
+ EventLoopGroup eventLoopGroup2 =
+ pulsarAdminImpl2.getAsyncHttpConnector().getHttpClient().getConfig().getEventLoopGroup();
+ Timer nettyTimer1 = pulsarAdminImpl1.getAsyncHttpConnector().getHttpClient().getConfig().getNettyTimer();
+ Timer nettyTimer2 = pulsarAdminImpl2.getAsyncHttpConnector().getHttpClient().getConfig().getNettyTimer();
+ boolean b1 = eventLoopGroup1 == eventLoopGroup2;
+ boolean b2 = nettyTimer1 == nettyTimer2;
+ assertThat(b1 && b2).isTrue();
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client]Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc commented on code in PR #24893:
URL: https://github.com/apache/pulsar/pull/24893#discussion_r2468542288
##########
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java:
##########
@@ -112,33 +120,57 @@ public AsyncHttpConnector(Client client, ClientConfigurationData conf, int autoC
(int) client.getConfiguration().getProperty(ClientProperties.READ_TIMEOUT),
PulsarAdminImpl.DEFAULT_REQUEST_TIMEOUT_SECONDS * 1000,
autoCertRefreshTimeSeconds,
- conf, acceptGzipCompression);
+ conf, acceptGzipCompression, null);
}
@SneakyThrows
public AsyncHttpConnector(int connectTimeoutMs, int readTimeoutMs,
int requestTimeoutMs,
int autoCertRefreshTimeSeconds, ClientConfigurationData conf,
- boolean acceptGzipCompression) {
+ boolean acceptGzipCompression,
+ PulsarClientSharedResourcesImpl sharedResources) {
Validate.notEmpty(conf.getServiceUrl(), "Service URL is not provided");
serviceNameResolver = new PulsarServiceNameResolver();
String serviceUrl = conf.getServiceUrl();
serviceNameResolver.updateServiceUrl(serviceUrl);
this.acceptGzipCompression = acceptGzipCompression;
AsyncHttpClientConfig asyncHttpClientConfig =
createAsyncHttpClientConfig(conf, connectTimeoutMs, readTimeoutMs, requestTimeoutMs,
- autoCertRefreshTimeSeconds);
+ autoCertRefreshTimeSeconds, sharedResources);
httpClient = createAsyncHttpClient(asyncHttpClientConfig);
this.requestTimeout = requestTimeoutMs > 0 ? Duration.ofMillis(requestTimeoutMs) : null;
this.maxRetries = httpClient.getConfig().getMaxRequestRetry();
+ this.nameResolver = buildNameResolverIfConfigured(sharedResources);
+ }
+ private NameResolver<InetAddress> buildNameResolverIfConfigured(PulsarClientSharedResourcesImpl sharedResources) {
+ if (sharedResources != null && sharedResources.getDnsResolverGroup() != null) {
+ EventLoopGroup eventLoopGroupReference;
+ if (sharedResources.getIoEventLoopGroup() != null) {
+ eventLoopGroupReference = sharedResources.getIoEventLoopGroup();
+ } else {
+ // build an EventLoopGroup with default value
+ eventLoopGroupReference = EventLoopUtil.newEventLoopGroup(
+ Runtime.getRuntime().availableProcessors(), false,
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-admin-client-io",
+ Thread.currentThread().isDaemon()));
Review Comment:
@lhotari Hi Lari, could you please review this when you have a moment? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client]Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "3pacccccc (via GitHub)" <gi...@apache.org>.
3pacccccc closed pull request #24893: [feat][client]Support shared resources in PulsarAdmin to reduce thread usage
URL: https://github.com/apache/pulsar/pull/24893
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-234: Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24893:
URL: https://github.com/apache/pulsar/pull/24893#issuecomment-3486382514
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24893?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `84.78261%` with `7 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 74.23%. Comparing base ([`676ba07`](https://app.codecov.io/gh/apache/pulsar/commit/676ba078de57ff1a8b30384babc88becdf61e566?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`9048d50`](https://app.codecov.io/gh/apache/pulsar/commit/9048d5049d1196ac51a4749abf6b0486ab5ffcb3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24893?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...client/admin/internal/http/AsyncHttpConnector.java](https://app.codecov.io/gh/apache/pulsar/pull/24893?src=pr&el=tree&filepath=pulsar-client-admin%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2Finternal%2Fhttp%2FAsyncHttpConnector.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9hZG1pbi9pbnRlcm5hbC9odHRwL0FzeW5jSHR0cENvbm5lY3Rvci5qYXZh) | 82.92% | [2 Missing and 5 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24893?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24893 +/- ##
=============================================
+ Coverage 38.56% 74.23% +35.67%
- Complexity 13262 33877 +20615
=============================================
Files 1856 1913 +57
Lines 145287 149488 +4201
Branches 16877 17371 +494
=============================================
+ Hits 56025 110973 +54948
+ Misses 81696 29647 -52049
- Partials 7566 8868 +1302
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24893/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24893/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.28% <43.47%> (+0.10%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24893/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `22.70% <43.47%> (-0.06%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24893/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.75% <84.78%> (+39.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24893?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [.../client/admin/internal/PulsarAdminBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24893?src=pr&el=tree&filepath=pulsar-client-admin%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2Finternal%2FPulsarAdminBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9hZG1pbi9pbnRlcm5hbC9QdWxzYXJBZG1pbkJ1aWxkZXJJbXBsLmphdmE=) | `76.19% <100.00%> (+43.18%)` | :arrow_up: |
| [.../pulsar/client/admin/internal/PulsarAdminImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24893?src=pr&el=tree&filepath=pulsar-client-admin%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2Finternal%2FPulsarAdminImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9hZG1pbi9pbnRlcm5hbC9QdWxzYXJBZG1pbkltcGwuamF2YQ==) | `79.04% <100.00%> (+10.47%)` | :arrow_up: |
| [...dmin/internal/http/AsyncHttpConnectorProvider.java](https://app.codecov.io/gh/apache/pulsar/pull/24893?src=pr&el=tree&filepath=pulsar-client-admin%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2Finternal%2Fhttp%2FAsyncHttpConnectorProvider.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9hZG1pbi9pbnRlcm5hbC9odHRwL0FzeW5jSHR0cENvbm5lY3RvclByb3ZpZGVyLmphdmE=) | `88.88% <ø> (ø)` | |
| [...client/admin/internal/http/AsyncHttpConnector.java](https://app.codecov.io/gh/apache/pulsar/pull/24893?src=pr&el=tree&filepath=pulsar-client-admin%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fadmin%2Finternal%2Fhttp%2FAsyncHttpConnector.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9hZG1pbi9pbnRlcm5hbC9odHRwL0FzeW5jSHR0cENvbm5lY3Rvci5qYXZh) | `85.08% <82.92%> (+11.72%)` | :arrow_up: |
... and [1407 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24893/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client] PIP-234: Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24893:
URL: https://github.com/apache/pulsar/pull/24893#discussion_r2480414235
##########
pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java:
##########
@@ -188,6 +195,96 @@ public void testClientDescription() throws PulsarClientException {
PulsarAdmin.builder().serviceHttpUrl("http://localhost:8080").description("forked").build();
}
+ @Test
+ public void testClientBuildWithSharedResources() throws PulsarClientException {
+ PulsarClientSharedResources sharedResources = PulsarClientSharedResources.builder()
+ .configureEventLoop(eventLoopGroupConfig -> {
+ eventLoopGroupConfig
+ .name("testEventLoop")
+ .numberOfThreads(20);
+ })
+ .configureDnsResolver(dnsResolverConfig -> {
+ dnsResolverConfig.localAddress(new InetSocketAddress(0));
+ })
+ .configureTimer(timerConfig -> {
+ timerConfig.name("testTimer").tickDuration(100, TimeUnit.MILLISECONDS);
+ })
+ .build();
+ // create two adminClients and check if they share the same event loop group and netty timer
+ @Cleanup
+ PulsarAdminImpl pulsarAdminImpl1 =
+ (PulsarAdminImpl) PulsarAdmin.builder()
+ .serviceHttpUrl("http://localhost:8080")
+ .sharedResources(sharedResources)
+ .build();
+ @Cleanup
+ PulsarAdminImpl pulsarAdminImpl2 =
+ (PulsarAdminImpl) PulsarAdmin.builder()
+ .serviceHttpUrl("http://localhost:8080")
+ .sharedResources(sharedResources)
+ .build();
+
+ EventLoopGroup eventLoopGroup1 =
+ pulsarAdminImpl1.getAsyncHttpConnector().getHttpClient().getConfig().getEventLoopGroup();
+ EventLoopGroup eventLoopGroup2 =
+ pulsarAdminImpl2.getAsyncHttpConnector().getHttpClient().getConfig().getEventLoopGroup();
+ Timer nettyTimer1 = pulsarAdminImpl1.getAsyncHttpConnector().getHttpClient().getConfig().getNettyTimer();
+ Timer nettyTimer2 = pulsarAdminImpl2.getAsyncHttpConnector().getHttpClient().getConfig().getNettyTimer();
+ boolean b1 = eventLoopGroup1 == eventLoopGroup2;
+ boolean b2 = nettyTimer1 == nettyTimer2;
+ assertThat(b1 && b2).isTrue();
Review Comment:
This is a good suggestion from Copilot. The goal of AssertJ is to provide proper error messages when something fails and the suggestion shows how AssertJ should be used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [feat][client]Support shared resources in PulsarAdmin to reduce thread usage [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24893:
URL: https://github.com/apache/pulsar/pull/24893#discussion_r2465741367
##########
pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java:
##########
@@ -112,33 +120,57 @@ public AsyncHttpConnector(Client client, ClientConfigurationData conf, int autoC
(int) client.getConfiguration().getProperty(ClientProperties.READ_TIMEOUT),
PulsarAdminImpl.DEFAULT_REQUEST_TIMEOUT_SECONDS * 1000,
autoCertRefreshTimeSeconds,
- conf, acceptGzipCompression);
+ conf, acceptGzipCompression, null);
}
@SneakyThrows
public AsyncHttpConnector(int connectTimeoutMs, int readTimeoutMs,
int requestTimeoutMs,
int autoCertRefreshTimeSeconds, ClientConfigurationData conf,
- boolean acceptGzipCompression) {
+ boolean acceptGzipCompression,
+ PulsarClientSharedResourcesImpl sharedResources) {
Validate.notEmpty(conf.getServiceUrl(), "Service URL is not provided");
serviceNameResolver = new PulsarServiceNameResolver();
String serviceUrl = conf.getServiceUrl();
serviceNameResolver.updateServiceUrl(serviceUrl);
this.acceptGzipCompression = acceptGzipCompression;
AsyncHttpClientConfig asyncHttpClientConfig =
createAsyncHttpClientConfig(conf, connectTimeoutMs, readTimeoutMs, requestTimeoutMs,
- autoCertRefreshTimeSeconds);
+ autoCertRefreshTimeSeconds, sharedResources);
httpClient = createAsyncHttpClient(asyncHttpClientConfig);
this.requestTimeout = requestTimeoutMs > 0 ? Duration.ofMillis(requestTimeoutMs) : null;
this.maxRetries = httpClient.getConfig().getMaxRequestRetry();
+ this.nameResolver = buildNameResolverIfConfigured(sharedResources);
+ }
+ private NameResolver<InetAddress> buildNameResolverIfConfigured(PulsarClientSharedResourcesImpl sharedResources) {
+ if (sharedResources != null && sharedResources.getDnsResolverGroup() != null) {
+ EventLoopGroup eventLoopGroupReference;
+ if (sharedResources.getIoEventLoopGroup() != null) {
+ eventLoopGroupReference = sharedResources.getIoEventLoopGroup();
+ } else {
+ // build an EventLoopGroup with default value
+ eventLoopGroupReference = EventLoopUtil.newEventLoopGroup(
+ Runtime.getRuntime().availableProcessors(), false,
+ new ExecutorProvider.ExtendedThreadFactory("pulsar-admin-client-io",
+ Thread.currentThread().isDaemon()));
Review Comment:
In this case the EventLoopGroup wouldn't get closed and it would leak resources. It would be better to move this logic earlier so that the created event loop group is also used for AsyncHttpClient in this case and it would be closed when the instance is closed.
Please also add a test for this case. Restricting sharing of resources can be achieved by using the `shareConfigured` method (org.apache.pulsar.client.api.PulsarClientSharedResourcesBuilder#shareConfigured ) so that the eventloopgroup isn't shared by not configuring it, but just calling `configureDnsResolver` to configure the shared dns resolver.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org