implement full retry of partial failures in firehose/streams

src/sinks/aws_kinesis/firehose/config.rs

@@ -193,7 +193,7 @@ impl RetryLogic for KinesisRetryLogic {
             let msg = format!("partial error count {}", response.failure_count);
             return RetryAction::Retry(msg.into());
         } else {
-            RetryAction::Successful
+            return RetryAction::Successful
         }
     }
 }

src/sinks/aws_kinesis/firehose/integration_tests.rs

@@ -57,6 +57,7 @@ async fn firehose_put_records() {
         tls: None,
         auth: Default::default(),
         acknowledgements: Default::default(),
+        request_retry_partial: Default::default(),
     };
 
     let config = KinesisFirehoseSinkConfig { batch, base };

src/sinks/aws_kinesis/firehose/record.rs

@@ -53,7 +53,7 @@ impl SendRecord for KinesisFirehoseClient {
         records: Vec<Self::T>,
         stream_name: String,
     ) -> Result<KinesisResponse, SdkError<Self::E>> {
-        let rec_count = records.len().clone();
+        let rec_count = records.len();
 
         self.client
             .put_record_batch()

src/sinks/aws_kinesis/streams/config.rs

@@ -206,7 +206,7 @@ impl RetryLogic for KinesisRetryLogic {
             let msg = format!("partial error count {}", response.failure_count);
             return RetryAction::Retry(msg.into());
         } else {
-            RetryAction::Successful
+            return RetryAction::Successful
         }
     }
 }

src/sinks/aws_kinesis/streams/integration_tests.rs

@@ -98,6 +98,7 @@ async fn kinesis_put_records_without_partition_key() {
         tls: Default::default(),
         auth: Default::default(),
         acknowledgements: Default::default(),
+        request_retry_partial: Default::default(),
     };
 
     let config = KinesisStreamsSinkConfig {

src/sinks/aws_kinesis/streams/record.rs

@@ -1,10 +1,9 @@
-use crate::sinks::aws_kinesis::KinesisResponse;
 use aws_sdk_kinesis::output::PutRecordsOutput;
 use aws_sdk_kinesis::types::{Blob, SdkError};
 use bytes::Bytes;
 use tracing::Instrument;
 
-use super::{KinesisClient, KinesisError, KinesisRecord, Record, SendRecord};
+use super::{KinesisClient, KinesisError, KinesisResponse, KinesisRecord, Record, SendRecord};
 
 #[derive(Clone)]
 pub struct KinesisStreamRecord {
@@ -69,7 +68,7 @@ impl SendRecord for KinesisStreamClient {
         records: Vec<Self::T>,
         stream_name: String,
     ) -> Result<KinesisResponse, SdkError<Self::E>> {
-        let rec_count = records.len().clone();
+        let rec_count = records.len();
         self.client
             .put_records()
             .set_records(Some(records))