Admin client `list_topics` is not asynchronous

[luis-nook]

Description

As opposed to most other methods in the AdminClient, the list_topics method does not return a future.

However, the c implementation does release the gil before the metadata call. It shouldn't be too difficult to implement a version that returns a future instead.

One problem here is that if for example the kafka cluster becomes unresponsive momentarily and a call to list_topics starts, the entire python program blocks AND cannot be timed out. Given that kafka never sops trying to connect, this can stall the program forever.

I'm happy to implement this change, say as an argument to list_topics to keep backwards compatibility:

await client.list_topics(as_async=True)

How to reproduce

• Create an admin client without starting a kafka cluster
• use the list_topics method in the admin client
• the program will block forever (or until the cluster is started and responds to the request)

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()): ('2.0.2', 33555199)
• Apache Kafka broker version:
• Client configuration: {...}
• Operating system: macOS monterrey 12.5.1
• Provide client logs (with 'debug': '..' as necessary)

%3|1678370935.960|FAIL|rdkafka#producer-1| [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: Connect to ipv4#127.0.0.1:29092 failed: Connection refused (after 2ms in state CONNECT)
^C%3|1678370936.146|FAIL|rdkafka#producer-1| [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: Connect to ipv6#[::1]:29092 failed: Connection refused (after 0ms in state CONNECT)
^C%3|1678370936.339|FAIL|rdkafka#producer-1| [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: Connect to ipv4#127.0.0.1:29092 failed: Connection refused (after 0ms in state CONNECT)
%3|1678370936.963|FAIL|rdkafka#producer-1| [thrd:localhost:29092/bootstrap]: localhost:29092/bootstrap: Connect to ipv6#[::1]:29092 failed: Connection refused (after 1ms in state CONNECT)

• Provide broker log excerpts
• Critical issue

[pranavrth]

We will discuss this internally and let you know.

[pranavrth]

I think the main concern in this issue is that the client blocks forever which is not correct. There is a timeout property present in the API which will timeout the request in the specified time if the response is not received. Please use that.

Refer doc