output/aws_dynamodb: mark unattempted chunks as failed on error

When a chunk fails and individual retries also fail, mark all items
in subsequent unattempted chunks as failed in the BatchError. This
prevents silent data loss where unattempted items would be incorrectly
treated as successful.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: twmb

internal/impl/aws/dynamodb/output.go

@@ -394,10 +394,7 @@ func (d *dynamoDBWriter) WriteBatch(ctx context.Context, b service.MessageBatch)
 	// DynamoDB BatchWriteItem API limit.
 	for chunkStart := 0; chunkStart < len(writeReqs); chunkStart += dynamoDBMaxBatchItems {
 		boff.Reset()
-		chunkEnd := chunkStart + dynamoDBMaxBatchItems
-		if chunkEnd > len(writeReqs) {
-			chunkEnd = len(writeReqs)
-		}
+		chunkEnd := min(chunkStart+dynamoDBMaxBatchItems, len(writeReqs))
 		chunk := writeReqs[chunkStart:chunkEnd]
 
 		batchResult, err := d.client.BatchWriteItem(ctx, &dynamodb.BatchWriteItemInput{
@@ -438,6 +435,11 @@ func (d *dynamoDBWriter) WriteBatch(ctx context.Context, b service.MessageBatch)
 				if batchErr.IndexedErrors() == 0 {
 					err = nil
 				} else {
+					// Mark all items in subsequent unattempted chunks as
+					// failed to prevent silent data loss.
+					for k := chunkEnd; k < len(writeReqs); k++ {
+						batchErr.Failed(k, headlineErr)
+					}
 					err = batchErr
 				}
 			}