Apache Pulsar Async Consumer Setup (Completable Future)

user_521990 0
Viewed 763

I am trying to setup an asynchronous consumer for Apache Pulsar, but my problem is only 1 message is received and no other messages come through unless I restart my Spring Boot application. Unfortunately, the documentation around using the CompletableFuture with Pulsar is not great and I am new to using them.

My code below:

@EventListener(ApplicationReadyEvent.class)
public void subscribe() throws PulsarClientException {

Consumer consumer = this.pulsarClient.newConsumer(JSONSchema.of(TenantMicroserviceProvisioningRequestSchema.class))
            .subscriptionName(this.pulsarSubscriptionName)
            .topic(this.pulsarTopicName)
            .ackTimeout(240, TimeUnit.SECONDS)
            .subscriptionType(SubscriptionType.Exclusive)
            .subscribe();

while(true) {
    CompletableFuture<Message> asyncMessage = consumer.receiveAsync();

asyncMessage.thenAcceptAsync(incomingMessage -> {
        TenantMicroserviceProvisioningRequestSchema schema = (TenantMicroserviceProvisioningRequestSchema) incomingMessage.getValue();
        
       LOGGER.info(String.format("***New provisioning request recieved for tenant database [%s] with instance code [%s]", schema.getDatabaseName(), schema.getInstanceCode()));
                
       consumer.acknowledgeAsync(incomingMessage.getMessageId());
       
    });
 }
}

In the Java Docs it does say "receiveAsync() should be called subsequently once returned CompletableFuture gets complete with received message. Else it creates backlog of receive requests in the application.", but I'm unsure of how to do this. I think this is causing the issue.

Docs: Pulsar Docs: https://pulsar.apache.org/docs/en/client-libraries-java/#async-receive Pulsar Java Doc for receieveAsync(): https://pulsar.apache.org/api/client/org/apache/pulsar/client/api/Consumer.html

*** UPDATE *** I've added the while loop, but when I do this my Spring Boot memory consumption floats up to 10 GB. Not sure, but wondering if it is how the future is setup.

completable-future apache-pulsar java
1 Answers

In order to prevent such excess memory consumption, you will want to add an intermediate data structure to your class that limits the number of outstanding CompletableFutures such as a LinkedBlockingQueue as shown below

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;

public class AsyncConsumerDemo {
    
    private static final LinkedBlockingQueue<CompletableFuture<Message<String>>> outstandingMessages = 
      new LinkedBlockingQueue<CompletableFuture<Message<String>>>(1000);

    private static final ExecutorService executor = Executors.newCachedThreadPool();

    public static void main(String[] args) throws PulsarClientException, InterruptedException, ExecutionException {
        
        PulsarClient pulsarClient = PulsarClient.builder()
                .serviceUrl("pulsar://broker:6650")
                .build();

        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                    .subscriptionName("pulsarSubscriptionName")
                    .topic("pulsarTopicName")
                    .ackTimeout(240, TimeUnit.SECONDS)
                    .subscriptionType(SubscriptionType.Shared)
                    .subscribe();
        
        new Thread(() -> { // Message Retrieval Thread
            CompletableFuture<Message<String>> future;
            
            while ( (future = consumer.receiveAsync()) != null) {
                outstandingMessages.put(future);
            }
          }).start();
        
        for (int numConsumers = 0; numConsumers < 10; numConsumers++) {
          executor.submit(() -> { // Message Consumer Thread
            while(true) {
                try {
                    CompletableFuture<Message<String>> future = outstandingMessages.take();
                    Message<String> msg = future.get();
                    
                    // Process the message
                    
                    consumer.acknowledgeAsync(msg.getMessageId());
                    
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }
          });
       }
    }
    
}
user_32293010
Related