impl nack consumer response

danube-client-proto/src/main/proto/DanubeApi.proto

@@ -63,6 +63,9 @@ service ConsumerService {
 
     // Acknowledges receipt of a message from the Consumer
     rpc Ack(AckRequest) returns (AckResponse);
+
+    // Negative acknowledgment for a message from the Consumer
+    rpc Nack(NackRequest) returns (NackResponse);
 }
 
 // Create Consumer request
@@ -136,6 +139,20 @@ message AckResponse {
     uint64 request_id = 1;
 }
 
+message NackRequest {
+    uint64 request_id = 1;
+    // Identifies the message, associated with a unique topic, subscription and the broker
+    MsgID msg_id = 2;
+    // Subscription name the consumer is subscribed to
+    string subscription_name = 3;
+    optional uint64 delay_ms = 4;
+    optional string reason = 5;
+}
+
+message NackResponse {
+    uint64 request_id = 1;
+}
+
 // ============================================================================================
 
 service Discovery {

danube-client/src/main/java/com/danubemessaging/client/Consumer.java

@@ -173,6 +173,44 @@ public void ack(StreamMessage message) {
         notifyMessageAcked(topicConsumer, message);
     }
 
+    /**
+     * Negatively acknowledges a message asynchronously.
+     *
+     * @param message the message to negatively acknowledge
+     * @param delayMs optional redelivery delay in milliseconds; may be null
+     * @param reason  optional reason for the nack; may be null
+     * @return a future that completes when the nack is sent
+     */
+    public CompletableFuture<Void> nackAsync(StreamMessage message, Long delayMs, String reason) {
+        return CompletableFuture.runAsync(() -> nack(message, delayMs, reason), client.ioExecutor());
+    }
+
+    /**
+     * Negatively acknowledges a message, signaling the broker that processing failed.
+     * The broker may redeliver the message depending on the subscription's failure policy.
+     *
+     * @param message the message to negatively acknowledge; must not be null
+     * @param delayMs optional redelivery delay in milliseconds; may be null
+     * @param reason  optional reason for the nack; may be null
+     * @throws com.danubemessaging.client.errors.DanubeClientException if the message's topic
+     *         has no associated consumer or the consumer is closed
+     */
+    public void nack(StreamMessage message, Long delayMs, String reason) {
+        ensureOpen();
+
+        if (message == null) {
+            throw new DanubeClientException("Message is required for nack");
+        }
+
+        TopicConsumer topicConsumer = consumerByTopic.get(message.messageId().topicName());
+        if (topicConsumer == null) {
+            throw new DanubeClientException(
+                    "No consumer found for topic in message id: " + message.messageId().topicName());
+        }
+
+        topicConsumer.nack(message, delayMs, reason);
+    }
+
     /**
      * Closes this consumer, cancels the receive loop, and releases all resources.
      * Idempotent — safe to call multiple times.

danube-client/src/main/java/com/danubemessaging/client/internal/consumer/TopicConsumer.java

@@ -229,6 +229,61 @@ public void ack(StreamMessage message) {
         }
     }
 
+    public void nack(StreamMessage message, Long delayMs, String reason) {
+        int attempts = 0;
+        while (true) {
+            ensureOpen();
+
+            if (consumerId == 0) {
+                throw new DanubeClientException("Consumer is not subscribed for topic: " + topic);
+            }
+
+            var connection = connectionManager.getConnection(brokerAddress.brokerUrl(), brokerAddress.connectUrl());
+            var stub = ConsumerServiceGrpc.newBlockingStub(connection.grpcChannel())
+                    .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata()));
+
+            var builder = DanubeApi.NackRequest.newBuilder()
+                    .setRequestId(requestId.incrementAndGet())
+                    .setMsgId(message.messageId().toProto())
+                    .setSubscriptionName(options.subscription());
+            if (delayMs != null) {
+                builder.setDelayMs(delayMs);
+            }
+            if (reason != null) {
+                builder.setReason(reason);
+            }
+            DanubeApi.NackRequest request = builder.build();
+
+            try {
+                stub.nack(request);
+                return;
+            } catch (RuntimeException error) {
+                if (retryManager.isUnrecoverable(error)) {
+                    notifyError(error, true);
+                    relookupAndResubscribe();
+                    attempts = 0;
+                    continue;
+                }
+
+                if (!retryManager.isRetryable(error)) {
+                    notifyError(error, false);
+                    throw new DanubeClientException("Failed to nack message for topic: " + topic, error);
+                }
+
+                notifyError(error, true);
+
+                attempts++;
+                if (attempts > retryManager.maxRetries()) {
+                    relookupAndResubscribe();
+                    attempts = 0;
+                    continue;
+                }
+
+                sleepBackoff(retryManager.calculateBackoff(attempts - 1));
+            }
+        }
+    }
+
     public synchronized void relookupAndResubscribe() {
         ensureOpen();
         cancelHealthCheckTask();

danube-client/src/test/java/com/danubemessaging/client/it/ReliableNackIT.java

@@ -0,0 +1,109 @@
+package com.danubemessaging.client.it;
+
+import com.danubemessaging.client.Consumer;
+import com.danubemessaging.client.DanubeClient;
+import com.danubemessaging.client.DispatchStrategy;
+import com.danubemessaging.client.Producer;
+import com.danubemessaging.client.SubType;
+import com.danubemessaging.client.model.StreamMessage;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static com.danubemessaging.client.it.TestHelpers.*;
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Reliable nack tests: verify that nacking a message on a reliable-dispatch
+ * topic causes the broker to redeliver the same message.
+ */
+class ReliableNackIT {
+
+    private void runReliableNack(String topicPrefix, SubType subType) throws Exception {
+        DanubeClient client = newClient();
+        String topic = uniqueTopic(topicPrefix);
+
+        Producer producer = client.newProducer()
+                .withTopic(topic)
+                .withName("producer_reliable_nack")
+                .withDispatchStrategy(DispatchStrategy.RELIABLE)
+                .build();
+        producer.create();
+
+        Consumer consumer = client.newConsumer()
+                .withTopic(topic)
+                .withConsumerName("cons_rel_nack")
+                .withSubscription("rel_sub_nack")
+                .withSubscriptionType(subType)
+                .build();
+        consumer.subscribe();
+
+        try {
+            byte[] payload = "nack-redelivery-test".getBytes();
+
+            AtomicReference<StreamMessage> firstDelivery = new AtomicReference<>();
+            AtomicReference<StreamMessage> redelivery = new AtomicReference<>();
+            AtomicReference<Throwable> error = new AtomicReference<>();
+            AtomicInteger deliveryCount = new AtomicInteger();
+            CountDownLatch done = new CountDownLatch(1);
+
+            consumer.receive().subscribe(new TestHelpers.MessageCollector(2) {
+                @Override
+                public void onNext(StreamMessage item) {
+                    try {
+                        int count = deliveryCount.incrementAndGet();
+                        if (count == 1) {
+                            // First delivery — nack it
+                            firstDelivery.set(item);
+                            assertArrayEquals(payload, item.payload(), "first delivery payload mismatch");
+                            consumer.nack(item, 0L, "testing nack redelivery");
+                        } else if (count == 2) {
+                            // Redelivered message — verify and ack
+                            redelivery.set(item);
+                            assertArrayEquals(payload, item.payload(), "redelivered payload mismatch");
+                            assertEquals(firstDelivery.get().messageId().topicOffset(),
+                                    item.messageId().topicOffset(),
+                                    "redelivered message should have same offset");
+                            consumer.ack(item);
+                            done.countDown();
+                        }
+                    } catch (Throwable t) {
+                        error.set(t);
+                        done.countDown();
+                    }
+                }
+            });
+
+            Thread.sleep(400);
+
+            producer.send(payload, Map.of());
+
+            assertTrue(done.await(15, TimeUnit.SECONDS),
+                    "Timeout: received " + deliveryCount.get() + "/2 deliveries");
+
+            if (error.get() != null) {
+                fail("Assertion failed in subscriber: " + error.get().getMessage());
+            }
+
+            assertNotNull(firstDelivery.get(), "first delivery should not be null");
+            assertNotNull(redelivery.get(), "redelivery should not be null");
+        } finally {
+            consumer.close();
+            client.close();
+        }
+    }
+
+    @Test
+    void reliableNackExclusive() throws Exception {
+        runReliableNack("/default/reliable_nack_exclusive", SubType.EXCLUSIVE);
+    }
+
+    @Test
+    void reliableNackShared() throws Exception {
+        runReliableNack("/default/reliable_nack_shared", SubType.SHARED);
+    }
+}

docker/danube_broker.yml

@@ -31,15 +31,15 @@ auto_create_topics: true
 # Security Configuration
 auth:
   mode: none # Options: none, tls, tlswithjwt
-  tls:
-    cert_file: "./cert/server-cert.pem"
-    key_file: "./cert/server-key.pem"
-    ca_file: "./cert/ca-cert.pem"
-    verify_client: false
-  jwt:
-    secret_key: "your-secret-key"
-    issuer: "danube-auth"
-    expiration_time: 3600 # in seconds
+#  tls:
+#    cert_file: "./cert/server-cert.pem"
+#    key_file: "./cert/server-key.pem"
+#    ca_file: "./cert/ca-cert.pem"
+#    verify_client: false
+#  jwt:
+#    secret_key: "your-secret-key"
+#    issuer: "danube-auth"
+#    expiration_time: 3600 # in seconds
 
 # Load Manager Configuration (Automated Proactive Rebalancing)
 load_manager: