The client reused in PulsarConsumerSource is closed after first failure. #6417
Description
Describe the bug
When flink restart PulsarConsumerSource, PulsarConsumerSource reuse client from cache, but the client is closed. then the AlreadyClosedException will be throwed again and again.
To Reproduce
start a PulsarConsumerSource in flink
make a exception or timeout
PulsarConsumerSource will restart again and again and throw AlreadyClosedException
Expected behavior
don't reuse cached client
Desktop (please complete the following information):
- OS: macos , pulsar 2.5.0 flink 1.10.0