refactor for new comments

vectordotdev · Jun 11, 2023 · 10be829 · 10be829
1 parent ecb27e5
commit 10be829
Show file tree

Hide file tree

Showing 5 changed files with 4 additions and 6 deletions.
diff --git a/src/sinks/aws_kinesis/config.rs b/src/sinks/aws_kinesis/config.rs
@@ -53,8 +53,6 @@ pub struct KinesisSinkBaseConfig {
     pub auth: AwsAuthentication,
 
     /// Whether or not to retry successful requests containing partial failures.
-    ///
-    /// Note: this can cause duplicates in Firehose.
     #[serde(default)]
     #[configurable(metadata(docs::advanced))]
     pub request_retry_partial: bool,

diff --git a/src/sinks/aws_kinesis/firehose/config.rs b/src/sinks/aws_kinesis/firehose/config.rs
@@ -188,7 +188,7 @@ impl RetryLogic for KinesisRetryLogic {
     }
 
     fn should_retry_response(&self, response: &Self::Response) -> RetryAction {
-        if self.retry_partial && response.failure_count > 0 {
+        if response.failure_count > 0 && self.retry_partial  {
             let msg = format!("partial error count {}", response.failure_count);
             RetryAction::Retry(msg.into())
         } else {

diff --git a/src/sinks/aws_kinesis/firehose/record.rs b/src/sinks/aws_kinesis/firehose/record.rs
@@ -56,7 +56,7 @@ impl SendRecord for KinesisFirehoseClient {
     ) -> Result<KinesisResponse, SdkError<Self::E>> {
         let rec_count = records.len();
         let total_size = records.iter().fold(0, |acc, record| {
-            acc + record.data().unwrap_or(&Blob::new(vec![])).as_ref().len()
+            acc + record.data().map(|v| v.as_ref().len()).unwrap_or_default()
         });
         self.client
             .put_record_batch()

diff --git a/src/sinks/aws_kinesis/streams/config.rs b/src/sinks/aws_kinesis/streams/config.rs
@@ -201,7 +201,7 @@ impl RetryLogic for KinesisRetryLogic {
     }
 
     fn should_retry_response(&self, response: &Self::Response) -> RetryAction {
-        if self.retry_partial && response.failure_count > 0 {
+        if response.failure_count > 0 && self.retry_partial {
             let msg = format!("partial error count {}", response.failure_count);
             RetryAction::Retry(msg.into())
         } else {

diff --git a/src/sinks/aws_kinesis/streams/record.rs b/src/sinks/aws_kinesis/streams/record.rs
@@ -72,7 +72,7 @@ impl SendRecord for KinesisStreamClient {
     ) -> Result<KinesisResponse, SdkError<Self::E>> {
         let rec_count = records.len();
         let total_size = records.iter().fold(0, |acc, record| {
-            acc + record.data().unwrap_or(&Blob::new(vec![])).as_ref().len()
+            acc + record.data().map(|v| v.as_ref().len()).unwrap_or_default()
         });
         self.client
             .put_records()