[python client] cannot close consumer after an unsubscribe

[Jaudouard]

Describe the bug
Currently, calling close() after unsubscribe() with the latest python client crashes. The connection with pulsar is not fully released if close() is not called, causing a leak in file descriptor.

To Reproduce
Steps to reproduce the behavior:

from pulsar import Client
client = Client(service_url='pulsar://localhost:6650')
consumer = client.subscribe("test_topic", subscription_name="Consumer 1")
consumer.unsubscribe()
consumer.close()

• The consumer.close() crashes and raises an AlreadyClosed: Pulsar error exception,
• Several file descriptor are left open because the connection wasn't fully closed.

full log:

2022-06-10 14:40:35.586 INFO  [139972470286144] Client:88 | Subscribing on Topic :test_topic
2022-06-10 14:40:35.586 INFO  [139972470286144] ClientConnection:189 | [<none> -> pulsar://localhost:6650] Create ClientConnection, timeout=10000
2022-06-10 14:40:35.586 INFO  [139972470286144] ConnectionPool:96 | Created connection for pulsar://localhost:6650
2022-06-10 14:40:35.587 INFO  [139972139529984] ClientConnection:375 | [[::1]:51540 -> [::1]:6650] Connected to broker
2022-06-10 14:40:35.589 INFO  [139972139529984] HandlerBase:64 | [persistent://public/default/test_topic, Consumer 1, 0] Getting connection from pool
2022-06-10 14:40:35.609 INFO  [139972139529984] ConsumerImpl:224 | [persistent://public/default/test_topic, Consumer 1, 0] Created consumer on broker [[::1]:51540 -> [::1]:6650] 
2022-06-10 14:40:35.609 INFO  [139972470286144] ConsumerImpl:275 | [persistent://public/default/test_topic, Consumer 1, 0] Unsubscribing
2022-06-10 14:40:35.612 INFO  [139972139529984] ConsumerImpl:308 | [persistent://public/default/test_topic, Consumer 1, 0] Unsubscribed successfully
---------------------------------------------------------------------------
AlreadyClosed                             Traceback (most recent call last)
Input In [12], in <cell line: 5>()
      3 consumer = client.subscribe("test_topic", subscription_name="Consumer 1")
      4 consumer.unsubscribe()
----> 5 consumer.close()

File ~/.virtualenvs/api_converter/lib/python3.8/site-packages/pulsar/__init__.py:1239, in Consumer.close(self)
   1235 def close(self):
   1236     """
   1237     Close the consumer.
   1238     """
-> 1239     self._consumer.close()
   1240     self._client._consumers.remove(self)

AlreadyClosed: Pulsar error: AlreadyClosed

Expected behavior
we can close the consumer after unsubcribing, releasing the file descriptors. This is the current normal behaviour in the Java client.

Desktop (please complete the following information):

• OS: Ubuntu 20.04
• python 3.8
• pulsar-client 2.10.0

Additional context
This issue follows a discussion with @BewareMyPower in apache/pulsar#14714

Thanks!

[github-actions]

The issue had no activity for 30 days, mark with Stale label.

[github-actions]

The issue had no activity for 30 days, mark with Stale label.