feat(kinesis): fix retry logic for request-level errors (#4167)

Author: jvh-aws

AmplifyClients/AmplifyKinesisClient/Sources/Support/RecordClient.swift

@@ -93,8 +93,8 @@ actor RecordClient {
   }
 
   private func handleFailedRequest(_ records: [Record]) async {
-    let retryable = records.filter { $0.retryCount + 1 < maxRetries }
-    let expired = records.filter { $0.retryCount + 1 >= maxRetries }
+    let retryable = records.filter { $0.retryCount < maxRetries }
+    let expired = records.filter { $0.retryCount >= maxRetries }
 
     do {
       try await storage.incrementRetryCount(ids: retryable.map(\.id))
@@ -104,7 +104,7 @@ actor RecordClient {
         let streamName = records[0].streamName
         logger.warn(
           "Deleted \(expired.count) records from stream \(streamName) "
-            + "that exceeded retry limit of \(maxRetries) after failed request"
+            + "that exceeded retry limit of \(maxRetries) after failed retries"
         )
       }
     } catch {

AmplifyClients/AmplifyKinesisClient/Tests/UnitTests/RecordClientFlushTests.swift

@@ -102,6 +102,9 @@ class RecordClientFlushTests: XCTestCase {
         let allRecords = try await storage.getRecordsByStream().flatMap { $0 }
         let record2Id = allRecords[1].id
         let record3Id = allRecords[2].id
+
+        // Increment to maxRetries (3) so they are expired on next failed flush
+        try await storage.incrementRetryCount(ids: [record2Id, record3Id])
         try await storage.incrementRetryCount(ids: [record2Id, record3Id])
         try await storage.incrementRetryCount(ids: [record2Id, record3Id])