fix: retry attempts calculation and logs in retry strategy (#2653)

Signed-off-by: adarsh0728 <gooneriitk@gmail.com>
Signed-off-by: Vigith Maurice <vigith@gmail.com>
Co-authored-by: Vigith Maurice <vigith@gmail.com>

Author: adarsh0728

rust/extns/numaflow-kafka/src/lib.rs

@@ -546,7 +546,7 @@ impl KafkaActor {
                     .fetch_metadata(Some(&topic), timeout)
                     .map_err(|e| Error::Kafka(format!("Failed to fetch metadata: {}", e)))?;
                 let Some(topic_metadata) = metadata.topics().first() else {
-                    warn!(topic = topic, "No topic metadata found");
+                    warn!(%topic, "No topic metadata found");
                     return Ok(0);
                 };
                 let mut topic_pending = 0;
@@ -605,7 +605,7 @@ impl KafkaActor {
                     .fetch_metadata(Some(&topic), timeout)
                     .map_err(|e| Error::Kafka(format!("Failed to fetch metadata: {}", e)))?;
                 let Some(topic_metadata) = metadata.topics().first() else {
-                    warn!(topic = topic, "No topic metadata found");
+                    warn!(%topic, "No topic metadata found");
                     return Ok(Vec::new());
                 };
                 let partitions: Vec<i32> =

rust/numaflow-core/src/pipeline/isb/jetstream/reader.rs

@@ -396,7 +396,7 @@ impl JetStreamReader {
                 let result = match msg.ack_with(ack_kind).await {
                     Ok(_) => Ok(()),
                     Err(e) => {
-                        warn!(?e, "Failed to send {:?} to Jetstream for message", ack_kind);
+                        warn!(error = ?e, ?ack_kind, "Failed to send ack to Jetstream for message");
                         Err(Error::Connection(format!(
                             "Failed to send {:?}: {:?}",
                             ack_kind, e

rust/numaflow-core/src/pipeline/isb/jetstream/writer.rs

@@ -342,7 +342,7 @@ impl JetstreamWriter {
                 Some(true) => {
                     // FIXME: add metrics
                     if log_counter >= 500 {
-                        warn!(stream=?stream, "stream is full (throttled logging)");
+                        warn!(?stream, "stream is full (throttled logging)");
                         log_counter = 0;
                     }
                     log_counter += 1;
@@ -439,8 +439,8 @@ impl JetstreamWriter {
                         if ack.duplicate {
                             warn!(
                                 message_id = ?message.id,
-                                stream = ?stream,
-                                ack = ?ack,
+                                ?stream,
+                                ?ack,
                                 "Duplicate message detected"
                             );
                         }

rust/numaflow-core/src/shared/grpc.rs

@@ -119,7 +119,7 @@ pub(crate) async fn create_rpc_channel(socket_path: PathBuf) -> error::Result<Ch
         async || match connect_with_uds(socket_path.clone()).await {
             Ok(channel) => Ok(channel),
             Err(e) => {
-                warn!(?e, ?socket_path, "Failed to connect to UDS socket");
+                warn!(error = ?e, ?socket_path, "Failed to connect to UDS socket");
                 Err(Error::Connection(format!(
                     "Failed to connect {socket_path:?}: {:?}",
                     e

rust/numaflow-core/src/shared/server_info.rs

@@ -261,8 +261,8 @@ fn check_sdk_compatibility(
     } else {
         // Language not found in the supported SDK versions
         warn!(
-            "SDK version constraint not found for language: {}, container type: {}",
-            sdk_language, container_type
+            %sdk_language,
+            %container_type, "SDK version constraint not found for language and container type"
         );
 
         // Return error indicating the language
@@ -407,12 +407,11 @@ async fn read_server_info(
                     // If the file ends with the END marker, trim it and break out of the loop
                     contents = data.trim_end_matches(END).to_string();
                     break;
-                } else {
-                    warn!("Server info file is incomplete, EOF is missing...");
                 }
+                warn!("Server info file is incomplete, EOF is missing...");
             }
             Err(e) => {
-                warn!("Failed to read file: {}", e);
+                warn!(error = ?e, ?file_path, "Failed to read server info file");
             }
         }
 

rust/numaflow-core/src/sink.rs

@@ -499,8 +499,8 @@ impl SinkWriter {
 
         let write_start_time = tokio::time::Instant::now();
         let total_msgs = messages.len();
+        let mut retry_attempts = 0;
         let total_msgs_bytes: usize = messages.iter().map(|msg| msg.value.len()).sum();
-        let mut attempts = 1;
         let mut error_map = HashMap::new();
         let mut fallback_msgs = Vec::new();
         let mut serving_msgs = Vec::new();
@@ -516,7 +516,7 @@ impl SinkWriter {
         let mut rng = StdRng::from_entropy();
 
         loop {
-            while attempts <= retry_config.sink_max_retry_attempts {
+            while retry_attempts <= retry_config.sink_max_retry_attempts {
                 let status = self
                     .write_to_sink_once(
                         &mut error_map,
@@ -528,11 +528,25 @@ impl SinkWriter {
                 match status {
                     Ok(true) => break,
                     Ok(false) => {
+                        // Increment retry attempt only if we will retry again
+                        // otherwise we will break out of the loop
+                        // sleep for the calculated delay only if we are retrying
+                        if retry_attempts >= retry_config.sink_max_retry_attempts {
+                            break;
+                        }
+                        retry_attempts += 1;
+                        write_errors_total += error_map.len();
                         warn!(
-                            "Retry attempt {} due to retryable error. Errors: {:?}",
-                            attempts, error_map
+                            retry_attempt = retry_attempts,
+                            ?error_map,
+                            "Retrying due to retryable error."
                         );
-                        write_errors_total += error_map.len();
+                        let delay = Self::calculate_exponential_delay(
+                            retry_config,
+                            retry_attempts,
+                            &mut rng,
+                        );
+                        sleep(Duration::from_millis(delay as u64)).await;
                     }
                     Err(e) => Err(e)?,
                 }
@@ -546,17 +560,11 @@ impl SinkWriter {
                             .to_string(),
                     ));
                 }
-
-                // Calculate exponential backoff delay for the next retry attempt
-                let delay = Self::calculate_exponential_delay(retry_config, attempts, &mut rng);
-                // Sleep for the calculated delay
-                sleep(Duration::from_millis(delay as u64)).await;
-                attempts += 1;
             }
 
             // If after the retries we still have messages to process, handle the post retry failures
             let need_retry = Self::handle_sink_post_retry(
-                &mut attempts,
+                &mut retry_attempts,
                 &mut error_map,
                 &mut fallback_msgs,
                 &mut messages_to_send,
@@ -566,9 +574,9 @@ impl SinkWriter {
             match need_retry {
                 // if we are done with the messages, break the loop
                 Ok(false) => break,
-                // if we need to retry, reset the attempts and error_map
+                // if we need to retry, reset the retry_attempts and error_map
                 Ok(true) => {
-                    attempts = 0;
+                    retry_attempts = 0;
                     error_map.clear();
                 }
                 Err(e) => Err(e)?,
@@ -642,7 +650,7 @@ impl SinkWriter {
     /// Handles the post retry failures based on the configured strategy,
     /// returns true if we need to retry, else false.
     fn handle_sink_post_retry(
-        attempts: &mut u16,
+        retry_attempts: &mut u16,
         error_map: &mut HashMap<String, i32>,
         fallback_msgs: &mut Vec<Message>,
         messages_to_send: &mut Vec<Message>,
@@ -658,17 +666,19 @@ impl SinkWriter {
             // if we need to retry, return true
             OnFailureStrategy::Retry => {
                 warn!(
-                    "Using onFailure Retry, Retry attempts {} completed",
-                    attempts
+                    retry_attempts = *retry_attempts,
+                    errors = ?error_map,
+                    "Using onFailure Retry, retry attempts completed"
                 );
                 return Ok(true);
             }
             // if we need to drop the messages, log and return false
             OnFailureStrategy::Drop => {
                 // log that we are dropping the messages as requested
                 warn!(
-                    "Dropping messages after {} attempts. Errors: {:?}",
-                    attempts, error_map
+                    retry_attempts = *retry_attempts,
+                    errors = ?error_map,
+                    "Dropping messages."
                 );
                 if is_mono_vertex() {
                     // update the drop metric count with the messages left for sink
@@ -693,8 +703,9 @@ impl SinkWriter {
             OnFailureStrategy::Fallback => {
                 // log that we are moving the messages to the fallback as requested
                 warn!(
-                    "Moving messages to fallback after {} attempts. Errors: {:?}",
-                    attempts, error_map
+                    retry_attempts = *retry_attempts,
+                    errors = ?error_map,
+                    "Moving messages to fallback after retry attempts.",
                 );
                 // move the messages to the fallback messages
                 fallback_msgs.append(messages_to_send);
@@ -772,7 +783,7 @@ impl SinkWriter {
             ));
         }
 
-        let mut attempts = 0;
+        let mut retry_attempts = 0;
         let mut fallback_error_map = HashMap::new();
         // start with the original set of message to be sent.
         // we will overwrite this vec with failed messages and will keep retrying.
@@ -783,10 +794,11 @@ impl SinkWriter {
             .clone()
             .backoff
             .unwrap();
-        let max_attempts = default_retry.steps.unwrap();
+        // steps is the maximum number of retry attempts, excluding the initial attempt
+        let max_retry_attempts = default_retry.steps.unwrap();
         let sleep_interval = default_retry.interval.unwrap();
 
-        while attempts < max_attempts {
+        while retry_attempts <= max_retry_attempts {
             match self.fb_sink(messages_to_send.clone()).await {
                 Ok(fb_response) => {
                     // create a map of id to result, since there is no strict requirement
@@ -856,26 +868,33 @@ impl SinkWriter {
                         ));
                     }
 
-                    attempts += 1;
-
                     if messages_to_send.is_empty() {
                         break;
                     }
 
-                    warn!(
-                        "Retry attempt {} due to retryable error. Errors: {:?}",
-                        attempts, fallback_error_map
-                    );
-                    sleep(tokio::time::Duration::from(sleep_interval)).await;
+                    // increment retry attempt only if we will retry
+                    // otherwise break out of the loop
+                    // sleep for the calculated delay only if we are retrying
+                    if retry_attempts < max_retry_attempts {
+                        retry_attempts += 1;
+                        warn!(
+                            retry_attempts,
+                            ?fallback_error_map,
+                            "Retrying due to retryable error in Fallback Sink"
+                        );
+                        sleep(tokio::time::Duration::from(sleep_interval)).await;
+                    } else {
+                        break;
+                    }
                 }
                 Err(e) => return Err(e),
             }
         }
         if !messages_to_send.is_empty() {
             return Err(Error::FbSink(format!(
-                "Failed to write messages to fallback sink after {} attempts. \
+                "Failed to write messages to fallback sink after {} retry attempts. \
                 Max Attempts configured: {} Errors: {:?}",
-                attempts, max_attempts, fallback_error_map
+                retry_attempts, max_retry_attempts, fallback_error_map
             )));
         }
         Ok(())
@@ -972,13 +991,15 @@ impl SinkWriter {
 
     fn calculate_exponential_delay(
         retry_config: &RetryConfig,
-        attempts: u16,
+        retry_attempts: u16,
         rng: &mut StdRng,
     ) -> f64 {
         // Calculate the base delay using the initial retry interval and the retry factor
-        // The base delay is calculated as: initial_retry_interval * retry_factor^(attempts-1)
+        // The base delay is calculated as: initial_retry_interval * retry_factor^(retry_attempts-1)
         let base_delay = (retry_config.sink_initial_retry_interval_in_ms as f64)
-            * retry_config.sink_retry_factor.powi((attempts - 1) as i32);
+            * retry_config
+                .sink_retry_factor
+                .powi((retry_attempts - 1) as i32);
 
         let jitter = retry_config.sink_retry_jitter;
         // If jitter is 0, return the base delay

rust/numaflow-core/src/source/generator.rs

@@ -78,17 +78,19 @@ mod stream_generator {
             let key_count = std::cmp::min(cfg.key_count as usize, cfg.rpu) as u8;
             if key_count != cfg.key_count {
                 warn!(
-                    "Specified KeyCount({}) is higher than RPU ({}). KeyCount is changed to {}",
-                    cfg.key_count, cfg.rpu, key_count
+                    key_count = cfg.key_count,
+                    rpu = cfg.rpu,
+                    new_key_count = key_count,
+                    "Specified KeyCount is higher than RPU. KeyCount has changed."
                 );
             }
             if key_count > 0 && rpu % key_count as usize != 0 {
                 let new_rpu = rpu - (rpu % key_count as usize);
                 warn!(
                     rpu,
                     key_count,
-                    "Specified RPU is not a multiple of the KeyCount. This may lead to uneven distribution of messages across keys. RPUs will be adjusted to {}",
-                    new_rpu
+                    new_rpu,
+                    "Specified RPU is not a multiple of the KeyCount. This may lead to uneven distribution of messages across keys. RPUs will be adjusted."
                 );
                 rpu = new_rpu;
             }

rust/numaflow-core/src/watermark/processor/manager.rs

@@ -226,7 +226,7 @@ impl ProcessorManager {
                 .await
                 .unwrap_or_else(|_| Vec::new()),
             Err(e) => {
-                warn!(?e, "Failed to get keys from ot bucket");
+                warn!(error = ?e, "Failed to get keys from ot bucket");
                 Vec::new()
             }
         };
@@ -303,7 +303,7 @@ impl ProcessorManager {
             let kv = match val {
                 Ok(kv) => kv,
                 Err(e) => {
-                    warn!(?e, "Failed to get next kv entry, recreating watcher");
+                    warn!(error = ?e, "Failed to get next kv entry, recreating watcher");
                     ot_watcher = Self::create_watcher(ot_bucket.clone()).await;
                     continue;
                 }
@@ -353,7 +353,7 @@ impl ProcessorManager {
             let kv = match val {
                 Ok(kv) => kv,
                 Err(e) => {
-                    warn!(?e, "Failed to get next kv entry, recreating watcher");
+                    warn!(error = ?e, "Failed to get next kv entry, recreating watcher");
                     hb_watcher = Self::create_watcher(hb_bucket.clone()).await;
                     continue;
                 }

rust/serving/src/app/orchestrator.rs

@@ -70,13 +70,13 @@ where
                     previous_pod_hash,
                 } => {
                     warn!(
-                        ?err,
+                        error = %err,
                         "Request was cancelled, fetching callbacks and responses again"
                     );
                     Some(previous_pod_hash)
                 }
                 status::Error::Duplicate(msg) => {
-                    warn!(error = ?msg, "Request already exists in the store");
+                    warn!(error = %msg, "Request already exists in the store");
                     serving_metrics().request_register_duplicate_count.inc();
                     return Err(Error::Duplicate(msg));
                 }
@@ -149,7 +149,8 @@ where
             }
             ProcessingStatus::Failed { error, pod_hash } => {
                 warn!(
-                    "Request was failed because of {error}, processing and fetching the responses again"
+                    %error,
+                    "Request was failed, processing and fetching the responses again"
                 );
                 let notify = self.process_request(id, request_type).await?;
                 notify.await.expect("sender was dropped")?;
@@ -184,13 +185,13 @@ where
                     previous_pod_hash,
                 } => {
                     warn!(
-                        ?err,
+                        error = %err,
                         "Request was cancelled, fetching callbacks and responses again"
                     );
                     Some(previous_pod_hash)
                 }
                 status::Error::Duplicate(msg) => {
-                    warn!(error = ?msg, "Request already exists in the tracker");
+                    warn!(error = %msg, "Request already exists in the tracker");
                     serving_metrics().request_register_duplicate_count.inc();
                     return Err(Error::Duplicate(msg));
                 }

rust/serving/src/app/store/cbstore/jetstream_store.rs

@@ -208,7 +208,10 @@ impl JetStreamCallbackStore {
                         Ok(callback_obj) => {
                             let callback = Arc::new(callback_obj);
                             if tx.send(callback).await.is_err() {
-                                warn!(request_id = ?id, "Receiver dropped during historical send. Stopping scan stream.");
+                                warn!(
+                                    request_id = %id,
+                                    "Receiver dropped during historical send. Stopping scan stream."
+                                );
                                 break;
                             }
                         }