chore: Fix kinesis retry logic for request-level errors (#3223)

Co-authored-by: Matt Creaser <mattwcc@amazon.com>
Co-authored-by: Abhash Kumar Singh <thisisabhash@gmail.com>

Author: 3 people

aws-kinesis/src/main/java/com/amplifyframework/recordcache/RecordClient.kt

@@ -80,7 +80,7 @@ internal class RecordClient(
 
     private suspend fun handleFailedRequest(records: List<Record>) {
         try {
-            val (recordsToRetry, recordsToDelete) = records.partition { it.retryCount + 1 < maxRetries }
+            val (recordsToRetry, recordsToDelete) = records.partition { it.retryCount < maxRetries }
             val recordIdsToIncrement = recordsToRetry.map { it.id }
             val recordIdsToDelete = recordsToDelete.map { it.id }
 
@@ -91,7 +91,7 @@ internal class RecordClient(
                 val streamName = records.first().streamName
                 logger.warn {
                     "Deleted ${recordIdsToDelete.size} records from stream $streamName " +
-                        "that exceeded retry limit of $maxRetries after failed request"
+                        "that exceeded retry limit of $maxRetries after failed retries"
                 }
             }
         } catch (storageError: Throwable) {

aws-kinesis/src/test/java/com/amplifyframework/recordcache/RecordClientFlushTest.kt

@@ -126,12 +126,12 @@ class RecordClientFlushTest {
         storage.addRecord(RecordInput(streamName, "key2", byteArrayOf(2))).getOrThrow()
         storage.addRecord(RecordInput(streamName, "key3", byteArrayOf(3))).getOrThrow()
 
-        // Set record 2 and 3 to retry count 2 (will be deleted on next failure since retryCount + 1 >= maxRetries)
+        // Set record 2 and 3 to retry count 3 (will be deleted on next failure since retryCount >= maxRetries)
         val allRecords = storage.getRecordsByStream().getOrThrow().flatten()
         val record2Id = allRecords[1].id
         val record3Id = allRecords[2].id
 
-        repeat(2) { storage.incrementRetryCount(listOf(record2Id, record3Id)).getOrThrow() }
+        repeat(3) { storage.incrementRetryCount(listOf(record2Id, record3Id)).getOrThrow() }
 
         // Configure mock sender to fail with a non-SDK error
         coEvery { mockSender.putRecords(streamName, any()) } returns