feat: add negative acknowledge support via async constructs

feat: add negative acknowledge support via async constructs

Author: nachatz

pulsar/asyncio.py

@@ -322,31 +322,28 @@ async def acknowledge_cumulative(
 
     async def negative_acknowledge(
         self,
-        message: Union[pulsar.Message, pulsar.MessageId,
-                       _pulsar.Message, _pulsar.MessageId]
+        message: Union[pulsar.Message, pulsar.MessageId, _pulsar.Message, _pulsar.MessageId]
     ) -> None:
         """
         Acknowledge the failure to process a single message asynchronously.
 
         When a message is "negatively acked" it will be marked for redelivery after
         some fixed delay. The delay is configurable when constructing the consumer
         with {@link ConsumerConfiguration#setNegativeAckRedeliveryDelayMs}.
-
         This call is not blocking.
-
+        
         Parameters
         ----------
-
         message:
             The received message or message id.
         """
-        future = asyncio.get_running_loop().create_future()
         if isinstance(message, pulsar.Message):
             msg = message._message
+        elif isinstance(message, pulsar.MessageId):
+            msg = message._msg_id
         else:
             msg = message
-        self._consumer.negative_acknowledge_async(msg, functools.partial(_set_future, future, value=None))
-        await future
+        await asyncio.to_thread(self._consumer.negative_acknowledge, msg)
 
     async def unsubscribe(self) -> None:
         """

src/consumer.cc

@@ -135,12 +135,12 @@ void Consumer_acknowledgeCumulativeAsync_message_id(Consumer& consumer, const Me
 
 void Consumer_negative_acknowledgeAsync(Consumer& consumer, const Message& msg, ResultCallback callback) {
     py::gil_scoped_release release;
-    consumer.negativeAcknowledgeAsync(msg, callback);
+    consumer.negativeAcknowledge(msg);
 }
 
 void Consumer_negative_acknowledgeAsync_message_id(Consumer& consumer, const MessageId& msgId, ResultCallback callback) {
     py::gil_scoped_release release;
-    consumer.negativeAcknowledgeAsync(msg, callback);
+    consumer.negativeAcknowledge(msgId);
 }
 
 void Consumer_closeAsync(Consumer& consumer, ResultCallback callback) {