feat: add negative ack redelivery delay and backoff support (PIP-106)

README.md

@@ -14,7 +14,8 @@ Features:
 - Automatic reconnection with exponential back off;
 - Message batching;
 - Compression with LZ4, zlib, zstd or Snappy (can be deactivated with Cargo features);
-- Telemetry using [tracing](https://github.com/tokio-rs/tracing) crate (can be activated with Cargo features).
+- Telemetry using [tracing](https://github.com/tokio-rs/tracing) crate (can be activated with Cargo features);
+- Configurable negative-acknowledgment redelivery delay with pluggable backoff policies
 
 ## Getting Started
 
@@ -31,6 +32,7 @@ Try out [examples](examples):
 - [producer](examples/producer.rs)
 - [consumer](examples/consumer.rs)
 - [reader](examples/reader.rs)
+- [negative_ack](examples/negative_ack.rs)
 
 ## Project Maintainers
 

examples/negative_ack.rs

@@ -0,0 +1,103 @@
+#[macro_use]
+extern crate serde;
+use std::{env, time::Duration};
+
+use futures::TryStreamExt;
+use pulsar::{
+    authentication::{basic::BasicAuthentication, oauth2::OAuth2Authentication},
+    Authentication, Consumer, DeserializeMessage, MultiplierRedeliveryBackoff, Payload, Pulsar,
+    SubType, TokioExecutor,
+};
+
+#[derive(Serialize, Deserialize)]
+struct TestData {
+    data: String,
+}
+
+impl DeserializeMessage for TestData {
+    type Output = Result<TestData, serde_json::Error>;
+
+    fn deserialize_message(payload: &Payload) -> Self::Output {
+        serde_json::from_slice(&payload.data)
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), pulsar::Error> {
+    env_logger::init();
+
+    let addr = env::var("PULSAR_ADDRESS")
+        .ok()
+        .unwrap_or_else(|| "pulsar://127.0.0.1:6650".to_string());
+    let topic = env::var("PULSAR_TOPIC")
+        .ok()
+        .unwrap_or_else(|| "non-persistent://public/default/test".to_string());
+
+    let mut builder = Pulsar::builder(addr, TokioExecutor);
+
+    if let Ok(token) = env::var("PULSAR_TOKEN") {
+        let authentication = Authentication {
+            name: "token".to_string(),
+            data: token.into_bytes(),
+        };
+
+        builder = builder.with_auth(authentication);
+    } else if let Ok(oauth2_cfg) = env::var("PULSAR_OAUTH2") {
+        builder = builder.with_auth_provider(OAuth2Authentication::client_credentials(
+            serde_json::from_str(oauth2_cfg.as_str())
+                .unwrap_or_else(|_| panic!("invalid oauth2 config [{}]", oauth2_cfg.as_str())),
+        ));
+    } else if let (Ok(username), Ok(password)) = (
+        env::var("PULSAR_BASIC_USERNAME"),
+        env::var("PULSAR_BASIC_PASSWORD"),
+    ) {
+        builder = builder.with_auth_provider(BasicAuthentication::new(&username, &password))
+    }
+
+    let pulsar: Pulsar<_> = builder.build().await?;
+    let negative_ack_backoff = MultiplierRedeliveryBackoff::builder()
+        .min_delay(Duration::from_secs(1))
+        .max_delay(Duration::from_secs(30))
+        .multiplier(2.0)
+        .build()?;
+
+    let mut consumer: Consumer<TestData, _> = pulsar
+        .consumer()
+        .with_topic(topic)
+        .with_consumer_name("negative_ack_consumer")
+        .with_subscription_type(SubType::Exclusive)
+        .with_subscription("negative_ack_subscription")
+        .with_nack_redelivery_delay(Duration::from_secs(5))
+        .with_negative_ack_backoff(negative_ack_backoff)
+        .build()
+        .await?;
+
+    let mut counter = 0usize;
+    while let Some(msg) = consumer.try_next().await? {
+        log::info!("metadata: {:?}", msg.metadata());
+        log::info!("id: {:?}", msg.message_id());
+        log::info!("redelivery_count: {}", msg.redelivery_count());
+
+        let data = match msg.deserialize() {
+            Ok(data) => data,
+            Err(e) => {
+                log::error!("could not deserialize message: {:?}", e);
+                consumer.nack(&msg).await?;
+                break;
+            }
+        };
+
+        log::info!("payload data: {}", data.data);
+        consumer.nack(&msg).await?;
+
+        counter += 1;
+        log::info!("nacked {} messages", counter);
+
+        if counter > 10 {
+            consumer.close().await.expect("Unable to close consumer");
+            break;
+        }
+    }
+
+    Ok(())
+}

src/consumer/builder.rs

@@ -1,6 +1,7 @@
 use std::{
     collections::{BTreeMap, VecDeque},
-    time::Duration,
+    sync::Arc,
+    time::{Duration, Instant},
 };
 
 use futures::{future::try_join_all, StreamExt};
@@ -10,13 +11,19 @@ use regex::Regex;
 use crate::{
     consumer::{
         config::ConsumerConfig, data::DeadLetterPolicy, multi::MultiTopicConsumer,
-        options::ConsumerOptions, topic::TopicConsumer, InnerConsumer,
+        negative_ack_backoff::NegativeAckBackoff, options::ConsumerOptions, topic::TopicConsumer,
+        InnerConsumer,
     },
     message::proto::command_subscribe::SubType,
     reader::{Reader, State},
     BrokerAddress, Consumer, DeserializeMessage, Error, Executor, Pulsar,
 };
 
+const NACK_REDELIVERY_DELAY_EXCEEDS_MILLISECONDS_RANGE: &str =
+    "negative-ack redelivery delay is too large: it must be less than u64::MAX milliseconds";
+const NACK_REDELIVERY_DELAY_EXCEEDS_INSTANT_RANGE: &str =
+    "negative-ack redelivery delay is too large to schedule from the current instant";
+
 /// Builder structure for consumers
 ///
 /// This is the main way to create a [Consumer] or a [Reader]
@@ -35,6 +42,24 @@ pub struct ConsumerBuilder<Exe: Executor> {
     consumer_options: Option<ConsumerOptions>,
     namespace: Option<String>,
     topic_refresh: Option<Duration>,
+    nack_redelivery_delay: Option<Duration>,
+    negative_ack_backoff: Option<Arc<dyn NegativeAckBackoff>>,
+}
+
+fn check_nack_delay_duration(delay: Duration) -> Result<(), Error> {
+    if delay.as_millis() >= u64::MAX as u128 {
+        return Err(Error::Custom(format!(
+            "{NACK_REDELIVERY_DELAY_EXCEEDS_MILLISECONDS_RANGE}: {delay:?}"
+        )));
+    }
+
+    if Instant::now().checked_add(delay).is_none() {
+        return Err(Error::Custom(format!(
+            "{NACK_REDELIVERY_DELAY_EXCEEDS_INSTANT_RANGE}: {delay:?}"
+        )));
+    }
+
+    Ok(())
 }
 
 impl<Exe: Executor> ConsumerBuilder<Exe> {
@@ -56,6 +81,8 @@ impl<Exe: Executor> ConsumerBuilder<Exe> {
             consumer_options: None,
             namespace: None,
             topic_refresh: None,
+            nack_redelivery_delay: None,
+            negative_ack_backoff: None,
         }
     }
 
@@ -175,6 +202,43 @@ impl<Exe: Executor> ConsumerBuilder<Exe> {
         self
     }
 
+    /// Sets the fixed delay before redelivering negatively acknowledged messages.
+    ///
+    /// If this option is not set, negative acknowledgments use a 60 seconds default delay before
+    /// broker redelivery.
+    ///
+    /// `Duration::ZERO` is valid and requests immediate redelivery. Repeated calls are
+    /// last-call-wins.
+    ///
+    /// The configured fixed delay is used for ID-only [`Self::nack_with_id`] calls, which carry no
+    /// redelivery count. It is also used as the delay for message-based [`Consumer::nack`] calls
+    /// when no backoff policy has been configured.
+    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
+    pub fn with_nack_redelivery_delay(mut self, delay: Duration) -> Self {
+        self.nack_redelivery_delay = Some(delay);
+        self
+    }
+
+    /// Configures an optional negative-ack backoff policy.
+    ///
+    /// Message-based negative acknowledgments use the message's redelivery count (`0` on first
+    /// delivery) to compute the backoff delay. ID-only `nack_with_id` calls have no redelivery
+    /// count, so they use
+    /// [`Self::with_nack_redelivery_delay`] or the 60 seconds default instead.
+    ///
+    /// A backoff that returns `Duration::ZERO` requests immediate redelivery.
+    ///
+    /// This policy affects only negative acknowledgments. Unacked-message redelivery and
+    /// dead-letter policy behavior remain configured separately.
+    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
+    pub fn with_negative_ack_backoff<B>(mut self, backoff: B) -> Self
+    where
+        B: NegativeAckBackoff + 'static,
+    {
+        self.negative_ack_backoff = Some(Arc::new(backoff) as Arc<dyn NegativeAckBackoff>);
+        self
+    }
+
     // Checks the builder for inconsistencies
     // returns a config and a list of topics with associated brokers
     #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
@@ -193,6 +257,8 @@ impl<Exe: Executor> ConsumerBuilder<Exe> {
             dead_letter_policy,
             namespace: _,
             topic_refresh: _,
+            nack_redelivery_delay,
+            negative_ack_backoff,
         } = self;
 
         if consumer_name.is_none() {
@@ -208,6 +274,9 @@ impl<Exe: Executor> ConsumerBuilder<Exe> {
                 "Cannot create consumer with no topics and no topic regex".into(),
             ));
         }
+        if let Some(delay) = nack_redelivery_delay {
+            check_nack_delay_duration(delay)?;
+        }
 
         let topics: Vec<(String, BrokerAddress)> = try_join_all(
             topics
@@ -263,6 +332,8 @@ impl<Exe: Executor> ConsumerBuilder<Exe> {
             unacked_message_redelivery_delay: unacked_message_resend_delay,
             options: consumer_options.unwrap_or_default(),
             dead_letter_policy,
+            nack_redelivery_delay,
+            negative_ack_backoff,
         };
         Ok((config, topics))
     }
@@ -347,3 +418,102 @@ impl<Exe: Executor> ConsumerBuilder<Exe> {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::{sync::Arc, time::Duration};
+
+    use super::*;
+
+    #[test]
+    fn test_nack_delay_validation_oversized() {
+        let delay = Duration::MAX;
+        let result = check_nack_delay_duration(delay);
+
+        assert_nack_delay_error(
+            result,
+            NACK_REDELIVERY_DELAY_EXCEEDS_MILLISECONDS_RANGE,
+            delay,
+        );
+    }
+
+    #[test]
+    fn test_nack_delay_validation_rejects_millisecond_range_boundary() {
+        let delay = Duration::from_millis(u64::MAX);
+        let result = check_nack_delay_duration(delay);
+
+        assert_nack_delay_error(
+            result,
+            NACK_REDELIVERY_DELAY_EXCEEDS_MILLISECONDS_RANGE,
+            delay,
+        );
+    }
+
+    fn assert_nack_delay_error(result: Result<(), Error>, expected_prefix: &str, delay: Duration) {
+        match result {
+            Err(Error::Custom(message)) => {
+                assert!(
+                    message.starts_with(expected_prefix),
+                    "expected error to start with {expected_prefix:?}, got {message:?}"
+                );
+                assert!(
+                    message.contains(&format!("{delay:?}")),
+                    "expected error to include the rejected Duration, got {message:?}"
+                );
+            }
+            other => panic!("expected Error::Custom({expected_prefix:?}), got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn test_nack_delay_validation_zero_is_valid() {
+        let result = check_nack_delay_duration(Duration::ZERO);
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_consumer_config_nack_delay_default_is_none() {
+        let config = ConsumerConfig::default();
+
+        assert!(config.nack_redelivery_delay.is_none());
+        assert!(config.negative_ack_backoff.is_none());
+    }
+
+    #[test]
+    fn test_consumer_config_stores_nack_delay() {
+        let config = ConsumerConfig {
+            nack_redelivery_delay: Some(Duration::from_secs(5)),
+            ..ConsumerConfig::default()
+        };
+
+        assert_eq!(config.nack_redelivery_delay, Some(Duration::from_secs(5)));
+    }
+
+    #[test]
+    fn test_consumer_config_stores_both_nack_configs_independently() {
+        let config = ConsumerConfig {
+            nack_redelivery_delay: Some(Duration::from_secs(5)),
+            negative_ack_backoff: Some(Arc::new(
+                crate::consumer::negative_ack_backoff::MultiplierRedeliveryBackoff::default(),
+            )),
+            ..ConsumerConfig::default()
+        };
+
+        assert!(config.nack_redelivery_delay.is_some());
+        assert!(config.negative_ack_backoff.is_some());
+    }
+
+    #[test]
+    fn test_consumer_config_backoff_only_stores_none_for_delay() {
+        let config = ConsumerConfig {
+            negative_ack_backoff: Some(Arc::new(
+                crate::consumer::negative_ack_backoff::MultiplierRedeliveryBackoff::default(),
+            )),
+            ..ConsumerConfig::default()
+        };
+
+        assert!(config.nack_redelivery_delay.is_none());
+        assert!(config.negative_ack_backoff.is_some());
+    }
+}