Pulsar client thread balance

I’m trying to implement a Pulsar client with multiple producers that distributes the load among the threads, but regardless the value passed on ioThreads() and on listenerThreads(), it is always overloading the first thread (> 65% cpu while the other threads are completely idle)

I have tried a few things including this “dynamic rebalancing” every hour(last method) but closing it in the middle of the process certainly is not the best approach

This is the relevant code

...
// pulsar client
pulsarClient = PulsarClient.builder() //
                           .operationTimeout(config.getAppPulsarTimeout(), TimeUnit.SECONDS) //
                           .ioThreads(config.getAppPulsarClientThreads()) //
                           .listenerThreads(config.getAppPulsarClientThreads()) //
                           .serviceUrl(config.getPulsarServiceUrl()).build();
...

private createProducers() {
    String strConsumerTopic = this.config.getPulsarTopicInput();
    List<Integer> protCasesList = this.config.getEventProtoCaseList();

    for (Integer e : protCasesList) {
        String topicName = config.getPulsarTopicOutput().concat(String.valueOf(e));
        LOG.info("Creating producer for topic: {}", topicName);

        Producer<byte[]> protobufProducer = pulsarClient.newProducer().topic(topicName).enableBatching(false)
                .blockIfQueueFull(true).compressionType(CompressionType.NONE)
                .sendTimeout(config.getPulsarSendTimeout(), TimeUnit.SECONDS)
                .maxPendingMessages(config.getPulsarMaxPendingMessages()).create();

        this.mapLink.put(strConsumerTopic.concat(String.valueOf(e)), protobufProducer);
    }
}

public void closeProducers() {
    String strConsumerTopic = this.config.getPulsarTopicInput();
    List<Integer> protCasesList = this.config.getEventProtoCaseList();

    for (Integer e : protCasesList) {
        try {
            this.mapLink.get(strConsumerTopic.concat(String.valueOf(e))).close();
            LOG.info("{} producer correctly closed...",
                    this.mapLink.get(strConsumerTopic.concat(String.valueOf(e))).getProducerName());
        } catch (PulsarClientException e1) {
            LOG.error("Producer: {} not closed cause: {}",
                    this.mapLink.get(strConsumerTopic.concat(String.valueOf(e))).getProducerName(),
                    e1.getMessage());
        }
    }
}

public void rebalancePulsarThreads(boolean firstRun) {
    ThreadMXBean threadHandler = ManagementFactory.getThreadMXBean();
    ThreadInfo[] threadsInfo = threadHandler.getThreadInfo(threadHandler.getAllThreadIds());
    for (ThreadInfo threadInfo : threadsInfo) {
        if (threadInfo.getThreadName().contains("pulsar-client-io")) {
            // enable cpu time for all threads
            threadHandler.setThreadCpuTimeEnabled(true);
            // get cpu time for this specific thread
            long threadCPUTime = threadHandler.getThreadCpuTime(threadInfo.getThreadId());
            int thresholdCPUTime = 65;
            if (threadCPUTime > thresholdCPUTime) {
                LOG.warn("Pulsar client thread with CPU time greater than {}% - REBALANCING now", thresholdCPUTime);
                try {
                    closeProducers();

                } catch (Exception e) {
                    if (!firstRun) {
                        // producers will not be available in the first run
                        // therefore, the logging only happens when it is not the first run
                        LOG.warn("Unable to close Pulsar client threads on rebalancing: {}", e.getMessage());
                    }
                }

                try {
                    createPulsarProducers();

                } catch (Exception e) {
                    LOG.warn("Unable to create Pulsar client threads on rebalancing: {}", e.getMessage());
                }
            }
        }
    }
}

Answer

From what you describe, the most likely scenario is that all the topics you’re using are served by one single broker.

If that’s indeed the case, and avoiding topic load balancing across brokers, it’s normal that it’s using a single thread because all these producers will be sharing a single, pooled, TCP connection and each connection is assigned to 1 IO thread (listener threads are used for consumer listeners).

If you want to force more threads, you can increase the “Max TCP connection per each broker” setting, in order to use all the configured IO threads.

eg:

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://localhost:6650")
    .ioThreads(16)
    .connectionsPerBroker(16)
    .create();