A proper recreation of the Producer

[bonza]

Hi,

I have a problem under the load when the Pulsar client's producer throws a timeout exception like this.
System.TimeoutException: Could not send message to broker within given timeout

I thought a recreation of the producer, in that case, might help.
I do it in the following order:

• Disposing of the old producer
• Closing the old producer's client
• Creating a new client
• Creating a new producer using the newly created client

But when I try to send the message using a new producer, I get Pulsar.Client.Api.AlreadyClosedException : producer(0, 1, -1) ConnectionHandleralready closed

What did I miss?

[Lanayx]

Hi! Several thoughts:

1. If you see "Could not send message to broker within given timeout", it means that the producer queue is full, which can be due to several reasons, like connection is down or server is down, or server is overloaded, or anything else, you should inspect producer logs for better understanding. If you think it's server overloaded, then producer reconnect will only worsen the situation
2. Normally you should never recreate producer (and especially PulsarClient instance), since it handles reconnects itself, dispose and close methods were made to handling the graceful shutdown
3. I can't imagine the situation where AlreadyClosedException is received on send, it should only be raised on producer disposal. Again, logs will be very helpful, I recommend this configuration
4. It's also worth mentioning that producer provides GetStats API which should be used with StatsInterval, that you can use to periodically check the pending queue length

[bonza]

Hi, and thank you for your thoughts!

the producer queue is full, which can be due to several reasons, like connection is down or server is down, or server is overloaded, or anything else

You are right! From my measurements, the producer's publish method takes around 30ms in my case, which means it can publish ~33 messages per second. However, the number of messages we need to process is much higher: 600 messages per second. This means that with one producer, its queue will be overloaded very quickly.
My solution is to have a pool of pulsar clients + producers (~20) and distribute the load across them based on the message hash. It almost works, but I occasionally get Timeout exceptions. Each timeout means waiting for the Web API call for customers, which is not cool.
My initial idea was to recreate the client and producer in case of a timeout and hope the new one would work better, but now it seems like a bad idea.

Now, I'll probably try to make the pool more "intelligent" and use GetStats API to find the best producer in the pool

[Lanayx]

However, the number of messages we need to process is much higher: 600 messages per second.

One Pulsar producer supports about 1M messages per second with the default configuration at my recent benchmark https://x.com/Lanayx/status/1852392154116555176. Are you using batching properly?

[bonza]

One Pulsar producer supports about 1M messages per second
I think it also depends on the message size, the serializer used, and, of course, whether the batching is enabled or not.
In my case, we use an AVRO serializer. I'm unsure about the AVRO message size, but JSON is about 2-5 Kb.
And no, batching is not enabled.
Enabling batching, from my understanding, means that the Web API returns a successful response as soon as the message is sent by the producer without waiting for confirmation of successful processing by Pulsar.
It also means that some messages could be silently lost if our Web API crashes, and I have a strict requirement not to lose any messages.

By the way, where should I look at if I have no stats returned?
Here is my code:

        var client = await new PulsarClientBuilder()
            .ServiceUrl(options.ServiceUrl)
            .StatsInterval(TimeSpan.FromSeconds(1))
            .Authentication(
                AuthenticationFactoryOAuth2.ClientCredentials(
                    issuerUrl,
                    options.Audience,
                    privateKeyDataUrl))
            .BuildAsync();

        var producer = await client
            .NewProducer(Schema.AVRO<T>())
            .ProducerName(producerName)
            .Topic(topic)
            .SendTimeout(TimeSpan.FromSeconds(sendTimeoutSeconds))
            .MaxPendingMessages(maxPendingMessages)
            .CreateAsync();

      var stats = await producer.GetStats();

[Lanayx]

Enabling batching, from my understanding, means that the Web API returns a successful response as soon as the message is sent by the producer without waiting for confirmation of successful processing by Pulsar.
It also means that some messages could be silently lost if our Web API crashes, and I have a strict requirement not to lose any messages.

No, when batching is enabled you don't have to lose messages, you just don't have to wait for the acknowledgement before sending other messages. For example, you have 100 concurrent requests and one shared producer, when calling SendAsync at the same time on the same producer, all those requests can be batched together in one message, and then each handler will await it's message acknowledgement separately.

By the way, where should I look at if I have no stats returned?

You can try running locally this integration test and compare with your current setup. Also, logs should be helpful, since stats are also logged

[bonza]

Thanks a lot, I found the issue in my code causing no stats being returned.