async/sync mode in pulsar producer and consumer

[ragaur-tibco]

What is the best way to check and confirm that Async/Sync mode is working in Pulsar producer and consumer

[lhotari]

What is the best way to check and confirm that Async/Sync mode is working in Pulsar producer and consumer

@ragaur-tibco what exactly do you mean with "async/sync mode working"? Please elaborate.

[ragaur-tibco]

@lhotari In Apache Pulsar, does async and sync mode work and how can I verify that the mode selected is working?

There are multiple APIs used by the producer to send messages.

1. send()

2. sendAsync()

While internally, TypedMessageBuilderImpl.class present in package org.apache.pulsar.client.impl send() method is calling sendAsync()

    public MessageId send() throws PulsarClientException {
        try {
            // enqueue the message to the buffer
            CompletableFuture<MessageId> sendFuture = sendAsync();

            if (!sendFuture.isDone()) {
                // the send request wasn't completed yet (e.g. not failing at enqueuing), then attempt to triggerFlush
                // it out
                producer.triggerFlush();
            }

            return sendFuture.get();
        } catch (Exception e) {
            throw PulsarClientException.unwrap(e);
        }
    }  

[ragaur-tibco]

@lhotari
Can you please confirm below points, In Pulsar Java

1. For synchronous mode we use send() method
2. For asynchronous mode we use sendAsync() method

if answer to the above two points is yes then why is send() method internally calling the sendAsync() method

[lhotari]

1. For synchronous mode we use send() method
2. For asynchronous mode we use sendAsync() method

@ragaur-tibco yes.

if answer to the above two points is yes then why is send() method internally calling the sendAsync() method

@ragaur-tibco I don't really understand the point of the question. Do you have a problem with this implementation? What are your expectations for send and sendAsync? Is there some expectation that isn't met?

With the current implementation in the Pulsar Java client, it simply fulfills the requirements when send() internally calls sendAsync() method. Perhaps you have different expectations or requirements.

[ragaur-tibco]

@lhotari I don't have any expectations or requirements I just want to know how to test the both modes any practical approach and anything in Pulsar SDK logs so that I can understand the internal process

[lhotari]

@ragaur-tibco For understanding the "internal process", it is good to start by reading the high level documentation for the Pulsar binary protocol. For example, the description of the producer gives good context to understand how the producer works. There's a presentation "Deep Dive into the Pulsar Binary Protocol" by @cbornet which explains a lot more around the binary protocol.
Pulsar's Java client and Pulsar broker use Netty to handle the network communications. If you really want to get into internal, it helps to get an understanding of Netty. One of the rare books on the topic is "Netty in Action".

I don't have any expectations or requirements I just want to know how to test the both modes any practical approach and anything in Pulsar SDK logs so that I can understand the internal process

As I have already mentioned in my previous comments, testing Pulsar is usually about validating some expectation, for example validating that a system meets both functional and non-functional requirements. There might be a test plan with test cases to make this well defined. Perhaps you are meaning more about experimentation when you say to "test the both modes"?
One possible approach is to look at the unit tests in the Pulsar source code and extract something out of that? For example, this test class contains a lot of different test cases.
With ChatGPT, you can also generate a lot of examples and just try to run them in the IDE with unit tests, if your main purpose is to experiment with things and learn Pulsar.
For Pulsar testing in general, you might get good ideas from some conference talks. I've done a talk "Validating Apache Pulsar’s Behavior Under Failure Conditions " on this subject, but it's a fairly high-level systems testing talk so it might not be practical for your case.

[ragaur-tibco]

@lhotari I have tried very simple implementation with pulsar as a way to identify the workings of sync and async modes which lets to no results.
From the sources jars I noticed that 'send()' method uses 'sendAsync()' method so can you please confirm with the 'send()' method that are we using sync mode or async mode

[visortelle]

@ragaur-tibco I read the whole discussion 3 times and am still unsure I understood your question.

I don't have any expectations or requirements I just want to know how to test the both modes any practical approach and anything in Pulsar SDK logs so that I can understand the internal process

@lhotari provided the file full of related test cases: https://github.com/apache/pulsar/blob/master/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java

I'm sure you'll find the answer to your question here.

Closing for now.

[lhotari]

I have tried very simple implementation with pulsar as a way to identify the workings of sync and async modes which lets to no results.

It might help to share more details about the problem.

From the sources jars I noticed that 'send()' method uses 'sendAsync()' method so can you please confirm with the 'send()' method that are we using sync mode or async mode

I can confirm your observation that send uses sendAsync in the implementation. When you are using "send", you are using the synchronous API call and when you are calling "sendAsync", it's an asynchronous API call, that returns a CompletableFuture. I hope this helps.

[lhotari]

When sending messages asynchronously, you cannot assume that they are sent from the client to the broker until the callback receives a message id from each message that has been sent. This comment about the correct usage of the Pulsar Java client might be helpful: #22011 (comment)

The relevant part:

The Pulsar client is designed to continue attempting to send messages until a potential send timeout occurs. It's also possible to set up an unlimited send timeout, allowing the client to retry indefinitely. This feature is detailed in the Pulsar documentation, available at https://pulsar.apache.org/docs/3.1.x/cookbooks-deduplication/#pulsar-clients (it's explained in the context of message deduplication). You can refer to the Javadocs for sendTimeout on ProducerBuilder.

It's crucial for messaging applications to be equipped to handle potential failures in message delivery, especially when data consistency is a key concern. Once the Pulsar client has acknowledged the message as sent by returning the message id, the responsibility for maintaining and ensuring the delivery of the message shifts to Pulsar. It's also necessary to verify that the message id is returned when using the asynchronous API (sendAsync).

If sending results in an error or the messaging application never receives a message id from the Pulsar client, it's the messaging application's responsibility to retry.