Waiting for consumers to complete task

[Tecno Brain <ce...@gmail.com>]

Hello,

 My application uses Pulsar as queues.  A Consumer will read a message from
topic T1, process it and write a message to topic T2. I have several
consumer threads on topic T1 (using a shared subscription).
...
    clientBuilder.listenerThreads(n);
    PulsarClient client = clientBuilder.build();


I am getting the following exception when my program terminates:

org.apache.pulsar.client.api.PulsarClientException$AlreadyClosedException:
Client already closed : state = Closing at
org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:975)
at
org.apache.pulsar.client.impl.ProducerBuilderImpl.create(ProducerBuilderImpl.java:91)
at com.example.messaging.broker.PulsarBroker.publish(PulsarBroker.java:94)

It seems that my application closes the PulsarClient after a thread had
read a message but before it writes to the second topic.

Is there a wait to avoid this issue?
If I call
  consumer.close();
in a Consumer<T> , would it wait for the thread to complete? Or will this
just avoid reading a new message?

-Jorge

Re: Waiting for consumers to complete task

[Tecno Brain <ce...@gmail.com>]

In other words, is it possible to stop the consumers from reading a new
message and wait for consumer threads already processing a message to
complete before closing the client?



On Tue, Jan 25, 2022 at 1:36 PM Tecno Brain <ce...@gmail.com>
wrote:

> Hello,
>
>  My application uses Pulsar as queues.  A Consumer will read a message
> from topic T1, process it and write a message to topic T2. I have several
> consumer threads on topic T1 (using a shared subscription).
> ...
>     clientBuilder.listenerThreads(n);
>     PulsarClient client = clientBuilder.build();
>
>
> I am getting the following exception when my program terminates:
>
> org.apache.pulsar.client.api.PulsarClientException$AlreadyClosedException:
> Client already closed : state = Closing at
> org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:975)
> at
> org.apache.pulsar.client.impl.ProducerBuilderImpl.create(ProducerBuilderImpl.java:91)
> at com.example.messaging.broker.PulsarBroker.publish(PulsarBroker.java:94)
>
> It seems that my application closes the PulsarClient after a thread had
> read a message but before it writes to the second topic.
>
> Is there a wait to avoid this issue?
> If I call
>   consumer.close();
> in a Consumer<T> , would it wait for the thread to complete? Or will this
> just avoid reading a new message?
>
> -Jorge
>
>
>