feat: add support to NegativeAcknowledge in Consumer

src/DotPulsar/Abstractions/IConsumer.cs

@@ -68,4 +68,19 @@ public interface IConsumer : IGetLastMessageIds, ISeek, IStateHolder<ConsumerSta
     /// Redeliver all pending messages that were pushed to this consumer that are not yet acknowledged.
     /// </summary>
     ValueTask RedeliverUnacknowledgedMessages(CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Acknowledge the consumption of a single message using the MessageId.
+    /// </summary>
+    /// <summary>
+    /// Negative acknowledge the consumption of a single message using the MessageId.
+    /// This will tell the broker the message was not processed successfully and should be redelivered later.
+    /// </summary>
+    ValueTask NegativeAcknowledge(MessageId messageId, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Negative acknowledge the consumption of multiple messages using the MessageIds.
+    /// This will tell the broker the messages were not processed successfully and should be redelivered later.
+    /// </summary>
+    ValueTask NegativeAcknowledge(IEnumerable<MessageId> messageIds, CancellationToken cancellationToken = default);
 }

src/DotPulsar/Extensions/ConsumerExtensions.cs

@@ -115,4 +115,11 @@ public static async ValueTask<ConsumerStateChanged> StateChangedFrom(this IConsu
         var currentState = await consumer.State.OnStateChangeFrom(state, delay, cancellationToken).ConfigureAwait(false);
         return new ConsumerStateChanged(consumer, currentState);
     }
+
+    /// <summary>
+    /// Negative acknowledge the consumption of a single message.
+    /// This will tell the broker the message was not processed successfully and should be redelivered later.
+    /// </summary>
+    public static async ValueTask NegativeAcknowledge(this IConsumer consumer, IMessage message, CancellationToken cancellationToken = default)
+        => await consumer.NegativeAcknowledge(message.MessageId, cancellationToken).ConfigureAwait(false);
 }

src/DotPulsar/Internal/Consumer.cs

@@ -420,6 +420,36 @@ public async ValueTask<IEnumerable<MessageId>> GetLastMessageIds(CancellationTok
         return messageIds;
     }
 
+    public async ValueTask NegativeAcknowledge(MessageId messageId, CancellationToken cancellationToken)
+    {
+        await Guard(cancellationToken).ConfigureAwait(false);
+
+        if (_singleSubConsumer is not null)
+            await _singleSubConsumer.NegativeAcknowledge(messageId, cancellationToken).ConfigureAwait(false);
+        else
+            await _subConsumers[messageId.Topic].NegativeAcknowledge(messageId, cancellationToken).ConfigureAwait(false);
+    }
+
+    public async ValueTask NegativeAcknowledge(IEnumerable<MessageId> messageIds, CancellationToken cancellationToken = default)
+    {
+        await Guard(cancellationToken).ConfigureAwait(false);
+
+        if (_singleSubConsumer is not null)
+        {
+            await _singleSubConsumer.NegativeAcknowledge(messageIds, cancellationToken).ConfigureAwait(false);
+            return;
+        }
+
+        var groupedMessageIds = messageIds.GroupBy(messageIds => messageIds.Topic);
+        var nackTasks = new List<Task>();
+        foreach (var group in groupedMessageIds)
+        {
+            nackTasks.Add(_subConsumers[group.Key].NegativeAcknowledge(group, cancellationToken).AsTask());
+        }
+
+        await Task.WhenAll(nackTasks).ConfigureAwait(false);
+    }
+
     private SubConsumer<TMessage> CreateSubConsumer(string topic)
     {
         var correlationId = Guid.NewGuid();

src/DotPulsar/Internal/SubConsumer.cs

@@ -237,4 +237,18 @@ private async ValueTask InternalAcknowledge(IEnumerable<MessageId> messageIds, C
 
     public ValueTask<IEnumerable<MessageId>> GetLastMessageIds(CancellationToken cancellationToken = default) =>
         throw new NotImplementedException();
+
+    public async ValueTask NegativeAcknowledge(MessageId messageId, CancellationToken cancellationToken)
+    {
+        var command = new CommandRedeliverUnacknowledgedMessages();
+        command.MessageIds.Add(messageId.ToMessageIdData());
+        await _executor.Execute(() => InternalRedeliverUnacknowledgedMessages(command, cancellationToken), cancellationToken).ConfigureAwait(false);
+    }
+
+    public async ValueTask NegativeAcknowledge(IEnumerable<MessageId> messageIds, CancellationToken cancellationToken = default)
+    {
+        var command = new CommandRedeliverUnacknowledgedMessages();
+        command.MessageIds.AddRange(messageIds.Select(messageId => messageId.ToMessageIdData()));
+        await _executor.Execute(() => InternalRedeliverUnacknowledgedMessages(command, cancellationToken), cancellationToken).ConfigureAwait(false);
+    }
 }

tests/DotPulsar.Tests/Internal/ConsumerTests.cs

@@ -467,6 +467,182 @@ public async Task Connectivity_WhenConnectionIsInitiallyUpAndReconnects_ShouldBe
         exception.ShouldBeNull();
     }
 
+    [Fact]
+    public async Task NegativeAcknowledge_GivenSingleMessage_ShouldBeRedelivered()
+    {
+        // Arrange
+        var topicName = await _fixture.CreateTopic(_cts.Token);
+        await using var client = CreateClient();
+        await using var consumer = CreateConsumer(client, topicName);
+        await using var producer = CreateProducer(client, topicName);
+
+        var content = "test-nack-message";
+        var messageId = await producer.Send(content, _cts.Token);
+
+        // Receive the message the first time
+        var message = await consumer.Receive(_cts.Token);
+        message.MessageId.ShouldBe(messageId);
+        message.Value().ShouldBe(content);
+
+        // Act
+        await consumer.NegativeAcknowledge(message.MessageId, _cts.Token);
+
+        // The message should be redelivered
+        var redeliveredMessage = await consumer.Receive(_cts.Token);
+
+        // Assert
+        redeliveredMessage.Value().ShouldBe(content);
+
+        // Verify message was redelivered (same content, may or may not have redelivery count)
+        // En algunas versiones de Pulsar, RedeliveryCount no se incrementa inmediatamente
+        // después de un negative acknowledgment
+        redeliveredMessage.Value().ShouldBe(message.Value());
+
+        // Clean up - acknowledge the message so it doesn't get redelivered further
+        await consumer.Acknowledge(redeliveredMessage.MessageId, _cts.Token);
+    }
+
+    [Fact]
+    public async Task NegativeAcknowledge_GivenMultipleMessages_ShouldAllBeRedelivered()
+    {
+        // Arrange
+        var topicName = await _fixture.CreateTopic(_cts.Token);
+        const int numberOfMessages = 5;
+
+        await using var client = CreateClient();
+        await using var consumer = CreateConsumer(client, topicName);
+        await using var producer = CreateProducer(client, topicName);
+
+        var messageIds = new List<MessageId>();
+        var contents = new List<string>();
+
+        // Produce and receive messages
+        for (var i = 0; i < numberOfMessages; i++)
+        {
+            var content = $"test-nack-message-{i}";
+            contents.Add(content);
+            messageIds.Add(await producer.Send(content, _cts.Token));
+        }
+
+        var receivedMessages = new List<IMessage<string>>();
+        for (var i = 0; i < numberOfMessages; i++)
+        {
+            receivedMessages.Add(await consumer.Receive(_cts.Token));
+        }
+
+        // Act
+        await consumer.NegativeAcknowledge(receivedMessages.Select(m => m.MessageId), _cts.Token);
+
+        // Assert
+        var redeliveredContents = new List<string>();
+        for (var i = 0; i < numberOfMessages; i++)
+        {
+            var redeliveredMessage = await consumer.Receive(_cts.Token);
+            redeliveredContents.Add(redeliveredMessage.Value());
+
+            // Acknowledge to prevent further redeliveries
+            await consumer.Acknowledge(redeliveredMessage.MessageId, _cts.Token);
+        }
+
+        // Verificamos que todos los contenidos originales se hayan reentregado
+        redeliveredContents.ShouldBe(contents, ignoreOrder: true);
+    }
+
+    [Fact]
+    public async Task NegativeAcknowledge_GivenPartitionedTopic_ShouldRedeliverMessagesFromAllPartitions()
+    {
+        // Arrange
+        const int numberOfMessages = 9;
+        const int partitions = 3;
+        var topicName = await _fixture.CreatePartitionedTopic(partitions, _cts.Token);
+
+        await using var client = CreateClient();
+        await using var consumer = CreateConsumer(client, topicName);
+        await using var producer = CreateProducer(client, topicName);
+
+        // Produce messages to ensure they spread across partitions
+        var messageContents = new List<string>();
+        for (var i = 0; i < numberOfMessages; i++)
+        {
+            var content = $"test-nack-message-{i}";
+            messageContents.Add(content);
+            await producer.Send(content, _cts.Token);
+        }
+
+        // Receive all messages
+        var receivedMessages = new List<IMessage<string>>();
+        for (var i = 0; i < numberOfMessages; i++)
+        {
+            receivedMessages.Add(await consumer.Receive(_cts.Token));
+        }
+
+        // Group messages by partition
+        var messagesByPartition = receivedMessages.GroupBy(m => m.MessageId.Partition)
+                                                .ToDictionary(g => g.Key, g => g.ToList());
+
+        // Verify we have messages from different partitions
+        messagesByPartition.Count.ShouldBeGreaterThan(1);
+
+        // Act - negative acknowledge all messages
+        await consumer.NegativeAcknowledge(receivedMessages.Select(m => m.MessageId), _cts.Token);
+
+        // Assert - all messages should be redelivered
+        var redeliveredContents = new List<string>();
+        var redeliveredPartitions = new HashSet<int>();
+
+        for (var i = 0; i < numberOfMessages; i++)
+        {
+            var redeliveredMessage = await consumer.Receive(_cts.Token);
+            redeliveredContents.Add(redeliveredMessage.Value());
+            redeliveredPartitions.Add(redeliveredMessage.MessageId.Partition);
+
+            // Acknowledge to prevent further redeliveries
+            await consumer.Acknowledge(redeliveredMessage.MessageId, _cts.Token);
+        }
+
+        // Verify all original messages were redelivered
+        redeliveredContents.ShouldBe(messageContents, ignoreOrder: true);
+
+        // Verify messages came from multiple partitions
+        redeliveredPartitions.Count.ShouldBeGreaterThan(1);
+    }
+
+    [Fact]
+    public async Task NegativeAcknowledge_WhenConnectionIsDisconnected_ShouldRedeliverAfterReconnect()
+    {
+        // Arrange
+        var topicName = await _fixture.CreateTopic(_cts.Token);
+        await using var client = CreateClient();
+        await using var consumer = CreateConsumer(client, topicName);
+        await using var producer = CreateProducer(client, topicName);
+
+        var messageContent = "test-nack-reconnect";
+
+        // Produce and consume a message
+        await producer.Send(messageContent, _cts.Token);
+        var message = await consumer.Receive(_cts.Token);
+        message.Value().ShouldBe(messageContent);
+
+        // Act - disconnect and nack the message
+        await using (await _fixture.DisableThePulsarConnection())
+        {
+            await consumer.StateChangedTo(ConsumerState.Disconnected, _cts.Token);
+        }
+
+        // Wait for reconnection
+        await consumer.StateChangedTo(ConsumerState.Active, _cts.Token);
+
+        // Now nack the message
+        await consumer.NegativeAcknowledge(message.MessageId, _cts.Token);
+
+        // Assert - should receive the message again
+        var redeliveredMessage = await consumer.Receive(_cts.Token);
+        redeliveredMessage.Value().ShouldBe(messageContent);
+
+        // Clean up
+        await consumer.Acknowledge(redeliveredMessage.MessageId, _cts.Token);
+    }
+
     private static async Task<IEnumerable<MessageId>> ProduceMessages(IProducer<string> producer, int numberOfMessages, string content, CancellationToken ct)
     {
         var messageIds = new MessageId[numberOfMessages];