Skip to content

Make the proxy adapter worker thread configrable #1549

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Technoboy- merged 1 commit into from
Nov 27, 2024
Merged

Make the proxy adapter worker thread configrable #1549

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions ...roxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,13 @@ public class MQTTProxyConfiguration extends MQTTCommonConfiguration {
)
private int maxNoOfChannels = 1;

@FieldContext(
category = CATEGORY_MQTT,
doc = "Number of threads to use for Netty IO for proxy to broker. "
+ "Default is set to `Runtime.getRuntime().availableProcessors()`"
)
private int proxyAdapterNumIOThreads = Runtime.getRuntime().availableProcessors();

@FieldContext(
category = CATEGORY_MQTT,
doc = "Proxy connect to broker timeout (ms)"
Expand Down
4 changes: 2 additions & 2 deletions ...xy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/channel/MQTTProxyAdapter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,14 +71,14 @@ public class MQTTProxyAdapter {
private final AtomicLong counter = new AtomicLong(0);
@Getter
private final ConcurrentMap<InetSocketAddress, Map<Integer, CompletableFuture<Channel>>> pool;
private final int workerThread = Runtime.getRuntime().availableProcessors();
private final int maxNoOfChannels;

public MQTTProxyAdapter(MQTTProxyService proxyService) {
this.proxyService = proxyService;
this.pool = new ConcurrentHashMap<>();
this.bootstrap = new Bootstrap();
this.eventLoopGroup = EventLoopUtil.newEventLoopGroup(workerThread, false, threadFactory);
this.eventLoopGroup = EventLoopUtil.newEventLoopGroup(
proxyService.getProxyConfig().getProxyAdapterNumIOThreads(), false, threadFactory);
this.maxNoOfChannels = proxyService.getProxyConfig().getMaxNoOfChannels();
this.bootstrap.group(eventLoopGroup)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, proxyService.getProxyConfig().getConnectTimeoutMs())
Expand Down
Loading