Why don't org.apache.pulsar.client.api.Producer and org.apache.pulsar.client.api.Consumer provide an isClose method?

[maozhitao12450]

We want to report client metrics to Prometheus.
Currently, we save a reference to the consumer/producer when it is created, and then we retrieve the metrics from org.apache.pulsar.client.api.Producer#getStats when updating the metrics.
However, in practice, some services use dynamically created and destroyed producers/consumers. Additionally, since the closing of producers and consumers is done during business operations, Pulsar does not provide any interceptors for closing them, and I cannot listen for their close events. Furthermore, pulsar do not provide an interface to check whether they are closed.
Therefore, could you provide an isClose interface?

[tisonkun]

Thanks for your reporting!

That sounds possible. What do you think @codelipenghui @eolivelli @BewareMyPower? If we have a lazy consensus here I can invest some time to try it out.

[BewareMyPower]

I cannot listen for their close events.

Since the close state can only be changed by calling closeAsync in your client side, you can remember the state by yourself.

If you want to mark it as closed when the state is Closing or Closed you can just write:

producer.closeAsync();
closed = true;

If you want to mark it as closed only when the state is Closed you can write:

producer.closeAsync().whenComplete((_, e) -> {
    if (e != null) {
        closed.set(true); // closed is an AtomicInteger
    }
});

If you even want to treat the Failed state as closed, you can write:

producer.closeAsync().whenComplete((_, e) -> {
    closed.set(true); // closed is an AtomicInteger
});

The current closeAsync API is flexible enough.

So The PulsarClient#isClosed does not make sense to me as well. @tisonkun

[tisonkun]

Make sense. After all in the reported case @maozhitao12450 has access to the producer. And you may wrap up your own producer extend:

public class ProducerExtend implements AutoClosable {
  public final Producer producer;
  public final AtomicBoolean closed = new AtomicBoolean();

  @Override
  public void close() throws Exception {
    // follow the comments above
  }
}

[maozhitao12450]

In fact, I thought about it yesterday and tried it. It worked, but there were loopholes. However, I think our service should be sufficient. But for the convenient of developers, I think a listener should still be added.

[maozhitao12450]

Today I tried to add wrapper to the producer.
Interestingly, if the producer is closed in a way other than producer.closeAsync();
I cannot know that it has been closed by listening to the AlreadyClosedException thrown by Producer#send. Actualy,send-action is trigger by TypedMessageBuilder

    MessageId messageId = producer.newMessage().value("Hello Pulsar").send();

I have to add another wrapper for TypedMessageBuilder, which is too troublesome.
If possible, a listener for the close event should be added or a simple status judgment (isClose) should be provided.
I communicated with Lari Hotari and felt that adding an event listener would be better.

[tisonkun]

Interestingly, if the producer is closed in a way other than producer.closeAsync();

Can you describe a bit concrete? I suppose a reasonable resource management code should handle liveness properly. Said:

try (producerExtend) {
  ...
}

// or propagate the lifecycle to any context outside like we combine ProducerExtend

public class XxxContext implements Closeable {
  @Override
  public void close() throws IOException {
    producerExtend.close();
  }
}

[tisonkun]

I have to add another wrapper for TypedMessageBuilder, which is too troublesome.

You mean you'd like to know if the producer is closed in the message build context? If so, maybe a demo repo can help. It more likes an application design issue.