[Bug] Acknowledge might fail with ResultAlreadyClosed during unload

[BewareMyPower]

Search before asking

• I searched in the issues and found nothing similar.

Version

main

Minimal reproduce step

Enable ackReceipt for https://github.com/apache/pulsar-client-cpp/blob/main/tests/extensibleLM/ExtensibleLoadManagerTest.cc

    ConsumerConfiguration consumerConf;
    consumerConf.setAckReceiptEnabled(true);
    Consumer consumer;
    Result consumerResult = client.subscribe(topicName, "sub", consumerConf, consumer);

What did you expect to see?

The test passed

What did you see instead?

2024-02-27 21:51:30.938 WARN  [0x16b36b000] AckGroupingTracker:93 | Connection is not ready, ACK failed for [(14,0,-1,-1)]
2024-02-27 21:51:30.938 INFO  [0x16b96f000] ExtensibleLoadManagerTest:145 | acked i: 0 81 ms
/Users/xuyunze/github.com/bewaremypower/pulsar-client-cpp/tests/extensibleLM/ExtensibleLoadManagerTest.cc:147: Failure
Expected equality of these values:
  ackResult
    Which is: AlreadyClosed
  ResultOk
    Which is: Ok

Note: I changed the log level of AckGroupingTracker:93 to warn locally.

Anything else?

When HandlerBase::getCnx() returns a null weak_ptr, it's retryable if the state is not Closing or Closed. We should schedule the flush again.

Are you willing to submit a PR?

• I'm willing to submit a PR!