use v2 API to create reader

Author: BewareMyPower

pulsar/asyncio.py

@@ -968,8 +968,8 @@ async def create_reader(self, topic: str,
         conf.crypto_failure_action(crypto_failure_action)
 
         future = asyncio.get_running_loop().create_future()
-        self._client.create_reader_async(
-            topic, start_message_id, conf, functools.partial(_set_future, future)
+        self._client.create_reader_async_v2(
+            topic, start_message_id, conf, functools.partial(_set_future_v2, future)
         )
         reader = await future
         schema.attach_client(self._client)

src/client.cc

@@ -124,6 +124,12 @@ void Client_createReaderAsync(Client& client, const std::string& topic, const Me
     client.createReaderAsync(topic, startMessageId, conf, callback);
 }
 
+void Client_createReaderAsyncV2(Client& client, const std::string& topic, const MessageId& startMessageId,
+                                ReaderConfiguration conf, ReaderV2Callback callback) {
+    py::gil_scoped_release release;
+    client.createReaderAsyncV2(topic, startMessageId, conf, std::move(callback));
+}
+
 std::vector<std::string> Client_getTopicPartitions(Client& client, const std::string& topic) {
     return waitForAsyncValue<std::vector<std::string>>(
         [&](GetPartitionsCallback callback) { client.getPartitionsForTopicAsync(topic, callback); });
@@ -211,6 +217,7 @@ void export_client(py::module_& m) {
         .def("subscribe_pattern", &Client_subscribe_pattern)
         .def("create_reader", &Client_createReader)
         .def("create_reader_async", &Client_createReaderAsync)
+        .def("create_reader_async_v2", &Client_createReaderAsyncV2)
         .def("create_table_view",
              [](Client& client, const std::string& topic, const TableViewConfiguration& config) {
                  return waitForAsyncValue<TableView>([&](TableViewCallback callback) {

tests/asyncio_test.py

@@ -597,6 +597,12 @@ def raise_exception():
         self.assertEqual(e.exception.error(), pulsar.Result.AuthenticationError)
         # TODO: we should fix the error message not included in pattern subscription case
 
+        with self.assertRaises(PulsarException) as e:
+            await client.create_reader("private/auth/asyncio-test-token-auth-reader",
+                                       pulsar.MessageId.earliest)
+        self.assertEqual(e.exception.error(), pulsar.Result.AuthenticationError)
+        self.assertIn("token supplier failed", str(e.exception))
+
         await client.close()