diff --git a/danube-client-proto/src/main/proto/DanubeApi.proto b/danube-client-proto/src/main/proto/DanubeApi.proto index 435ae8b..13e6eb4 100644 --- a/danube-client-proto/src/main/proto/DanubeApi.proto +++ b/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 { diff --git a/danube-client/src/main/java/com/danubemessaging/client/Consumer.java b/danube-client/src/main/java/com/danubemessaging/client/Consumer.java index da7f353..5b8d312 100644 --- a/danube-client/src/main/java/com/danubemessaging/client/Consumer.java +++ b/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 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. diff --git a/danube-client/src/main/java/com/danubemessaging/client/internal/consumer/TopicConsumer.java b/danube-client/src/main/java/com/danubemessaging/client/internal/consumer/TopicConsumer.java index 69d19a8..cd05be2 100644 --- a/danube-client/src/main/java/com/danubemessaging/client/internal/consumer/TopicConsumer.java +++ b/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(); diff --git a/danube-client/src/test/java/com/danubemessaging/client/it/ReliableNackIT.java b/danube-client/src/test/java/com/danubemessaging/client/it/ReliableNackIT.java new file mode 100644 index 0000000..614d3ad --- /dev/null +++ b/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 firstDelivery = new AtomicReference<>(); + AtomicReference redelivery = new AtomicReference<>(); + AtomicReference 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); + } +} diff --git a/docker/danube_broker.yml b/docker/danube_broker.yml index 9274318..6c58672 100644 --- a/docker/danube_broker.yml +++ b/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: