[receiver/awss3receiver] Fix data loss on partial SQS message processing failure

.chloggen/awss3receiver-fix-sqs-partial-failure.yaml

@@ -0,0 +1,32 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: receiver/awss3
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix data loss when SQS messages contain multiple S3 object notifications and some fail to process
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [45153]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  The SQS notification reader was unconditionally deleting messages after processing,
+  even when some S3 object retrievals or callback processing failed. This caused data
+  loss when a message contained multiple S3 notification records and any of them failed.
+  Messages are now only deleted when all records are successfully processed, allowing
+  failed records to be retried after the visibility timeout.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/awss3receiver/s3sqsreader.go

@@ -162,6 +162,10 @@ func (r *s3SQSNotificationReader) readAll(ctx context.Context, _ string, callbac
 					}
 				}
 
+				// Track whether all records were successfully processed.
+				// Only delete the message if all records succeed to prevent data loss.
+				allRecordsSucceeded := true
+
 				// Process each S3 object notification
 				for _, record := range s3Event.Records {
 					if record.EventSource != "aws:s3" || !strings.HasPrefix(record.EventName, "ObjectCreated:") {
@@ -202,6 +206,7 @@ func (r *s3SQSNotificationReader) readAll(ctx context.Context, _ string, callbac
 							zap.String("bucket", bucket),
 							zap.String("key", decodedKey),
 							zap.Error(err))
+						allRecordsSucceeded = false
 						continue
 					}
 
@@ -210,15 +215,24 @@ func (r *s3SQSNotificationReader) readAll(ctx context.Context, _ string, callbac
 						r.logger.Error("Failed to process S3 object content",
 							zap.String("key", decodedKey),
 							zap.Error(err))
+						allRecordsSucceeded = false
+						continue
 					}
 				}
 
-				_, err = r.sqsClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{
-					QueueUrl:      aws.String(r.queueURL),
-					ReceiptHandle: message.ReceiptHandle,
-				})
-				if err != nil {
-					r.logger.Warn("Failed to delete message from SQS queue", zap.Error(err))
+				// Only delete the message if all records were successfully processed.
+				// If any record failed, leave the message in the queue for retry.
+				if allRecordsSucceeded {
+					_, err = r.sqsClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{
+						QueueUrl:      aws.String(r.queueURL),
+						ReceiptHandle: message.ReceiptHandle,
+					})
+					if err != nil {
+						r.logger.Warn("Failed to delete message from SQS queue", zap.Error(err))
+					}
+				} else {
+					r.logger.Warn("Message not deleted due to processing failures, will be retried after visibility timeout",
+						zap.String("receiptHandle", *message.ReceiptHandle))
 				}
 			}
 		}

receiver/awss3receiver/s3sqsreader_test.go

@@ -432,7 +432,7 @@ func TestS3SQSReader_ReadAllErrorHandling(t *testing.T) {
 		assert.Equal(t, context.Canceled, err)
 	})
 
-	t.Run("handles S3 object retrieval error", func(t *testing.T) {
+	t.Run("does not delete message on S3 retrieval error", func(t *testing.T) {
 		mockS3 := new(mockS3ClientSQS)
 		mockSQS := new(mockSQSClient)
 
@@ -506,23 +506,216 @@ func TestS3SQSReader_ReadAllErrorHandling(t *testing.T) {
 			errors.New("object retrieval failed"),
 		)
 
-		// Mock message deletion
-		mockSQS.On("DeleteMessage", mock.Anything, mock.MatchedBy(func(input *sqs.DeleteMessageInput) bool {
-			return *input.QueueUrl == cfg.SQS.QueueURL &&
-				*input.ReceiptHandle == "test-receipt-handle"
-		})).Return(
-			&sqs.DeleteMessageOutput{},
+		// NOTE: DeleteMessage should NOT be called when S3 retrieval fails
+		// The message should remain in the queue for retry
+
+		ctx, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond)
+		defer cancel()
+		err = reader.readAll(ctx, "test-telemetry", func(context.Context, string, []byte) error {
+			t.Fatal("Callback should not be called when S3 retrieval fails")
+			return nil
+		})
+		assert.Error(t, err)
+		mockS3.AssertExpectations(t)
+		mockSQS.AssertExpectations(t)
+		mockSQS.AssertNotCalled(t, "DeleteMessage", mock.Anything, mock.Anything)
+	})
+
+	t.Run("does not delete message on partial failure", func(t *testing.T) {
+		mockS3 := new(mockS3ClientSQS)
+		mockSQS := new(mockSQSClient)
+
+		reader := &s3SQSNotificationReader{
+			logger:              logger,
+			s3Client:            mockS3,
+			sqsClient:           mockSQS,
+			queueURL:            cfg.SQS.QueueURL,
+			s3Bucket:            cfg.S3Downloader.S3Bucket,
+			s3Prefix:            cfg.S3Downloader.S3Prefix,
+			maxNumberOfMessages: 10,
+			waitTimeSeconds:     20,
+		}
+
+		// Create S3 event notification with THREE objects:
+		// - First will succeed
+		// - Second will fail S3 retrieval
+		// - Third will fail callback processing
+		s3Event := s3EventNotification{
+			Records: []s3EventRecord{
+				{
+					EventSource: "aws:s3",
+					EventName:   "ObjectCreated:Put",
+					S3: s3Data{
+						Bucket: s3BucketData{Name: "test-bucket"},
+						Object: s3ObjectData{Key: "success-key"},
+					},
+				},
+				{
+					EventSource: "aws:s3",
+					EventName:   "ObjectCreated:Put",
+					S3: s3Data{
+						Bucket: s3BucketData{Name: "test-bucket"},
+						Object: s3ObjectData{Key: "s3-failure-key"},
+					},
+				},
+				{
+					EventSource: "aws:s3",
+					EventName:   "ObjectCreated:Put",
+					S3: s3Data{
+						Bucket: s3BucketData{Name: "test-bucket"},
+						Object: s3ObjectData{Key: "callback-failure-key"},
+					},
+				},
+			},
+		}
+
+		eventJSON, err := json.Marshal(s3Event)
+		require.NoError(t, err)
+
+		mockSQS.On("ReceiveMessage", mock.Anything, mock.Anything).Return(
+			&sqs.ReceiveMessageOutput{
+				Messages: []types.Message{
+					{
+						Body:          aws.String(string(eventJSON)),
+						ReceiptHandle: aws.String("test-receipt-handle"),
+					},
+				},
+			},
+			nil,
+		).Once()
+
+		// After processing one message, return empty results to exit the loop.
+		mockSQS.On("ReceiveMessage", mock.Anything, mock.Anything).Return(
+			&sqs.ReceiveMessageOutput{
+				Messages: []types.Message{},
+			},
 			nil,
 		)
 
+		// First S3 object succeeds.
+		mockS3.On("GetObject", mock.Anything, &s3.GetObjectInput{
+			Bucket: aws.String("test-bucket"),
+			Key:    aws.String("success-key"),
+		}).Return([]byte("success-content"), nil)
+
+		// Second S3 object FAILS to retrieve.
+		mockS3.On("GetObject", mock.Anything, &s3.GetObjectInput{
+			Bucket: aws.String("test-bucket"),
+			Key:    aws.String("s3-failure-key"),
+		}).Return([]byte{}, errors.New("S3 GetObject failed"))
+
+		// Third S3 object retrieves successfully.
+		mockS3.On("GetObject", mock.Anything, &s3.GetObjectInput{
+			Bucket: aws.String("test-bucket"),
+			Key:    aws.String("callback-failure-key"),
+		}).Return([]byte("callback-content"), nil)
+
+		// NOTE: DeleteMessage should NOT be called when any record fails.
+		// The message should remain in the queue for retry.
+
 		ctx, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond)
 		defer cancel()
-		err = reader.readAll(ctx, "test-telemetry", func(_ context.Context, _ string, _ []byte) error {
-			t.Fatal("Callback should not be called when S3 retrieval fails")
+
+		successfulCallbacks := 0
+		failedCallbacks := 0
+
+		err = reader.readAll(ctx, "test-telemetry", func(_ context.Context, key string, _ []byte) error {
+			if key == "callback-failure-key" {
+				failedCallbacks++
+				return errors.New("callback processing failed")
+			}
+			successfulCallbacks++
 			return nil
 		})
 
-		assert.Error(t, err)
+		assert.Error(t, err) // Context timeout expected.
+
+		// Verify that only 1 out of 3 objects was successfully processed.
+		assert.Equal(t, 1, successfulCallbacks, "Only 1 object should have been successfully processed")
+		assert.Equal(t, 1, failedCallbacks, "1 object should have had callback failure")
+
+		mockS3.AssertExpectations(t)
+		mockSQS.AssertExpectations(t)
+		// Verify DeleteMessage was never called - message should remain for retry.
+		mockSQS.AssertNotCalled(t, "DeleteMessage", mock.Anything, mock.Anything)
+	})
+
+	t.Run("does not delete message on callback error", func(t *testing.T) {
+		mockS3 := new(mockS3ClientSQS)
+		mockSQS := new(mockSQSClient)
+
+		reader := &s3SQSNotificationReader{
+			logger:              logger,
+			s3Client:            mockS3,
+			sqsClient:           mockSQS,
+			queueURL:            cfg.SQS.QueueURL,
+			s3Bucket:            cfg.S3Downloader.S3Bucket,
+			s3Prefix:            cfg.S3Downloader.S3Prefix,
+			maxNumberOfMessages: 10,
+			waitTimeSeconds:     20,
+		}
+
+		s3Event := s3EventNotification{
+			Records: []s3EventRecord{
+				{
+					EventSource: "aws:s3",
+					EventName:   "ObjectCreated:Put",
+					S3: s3Data{
+						Bucket: s3BucketData{Name: "test-bucket"},
+						Object: s3ObjectData{Key: "test-key"},
+					},
+				},
+			},
+		}
+
+		eventJSON, err := json.Marshal(s3Event)
+		require.NoError(t, err)
+
+		mockSQS.On("ReceiveMessage", mock.Anything, mock.Anything).Return(
+			&sqs.ReceiveMessageOutput{
+				Messages: []types.Message{
+					{
+						Body:          aws.String(string(eventJSON)),
+						ReceiptHandle: aws.String("test-receipt-handle"),
+					},
+				},
+			},
+			nil,
+		).Once()
+
+		// After processing one message, return empty results to exit the loop.
+		mockSQS.On("ReceiveMessage", mock.Anything, mock.Anything).Return(
+			&sqs.ReceiveMessageOutput{
+				Messages: []types.Message{},
+			},
+			nil,
+		)
+
+		// S3 object retrieves successfully.
+		mockS3.On("GetObject", mock.Anything, &s3.GetObjectInput{
+			Bucket: aws.String("test-bucket"),
+			Key:    aws.String("test-key"),
+		}).Return([]byte("test-content"), nil)
+
+		// NOTE: DeleteMessage should NOT be called when callback fails.
+		// The message should remain in the queue for retry.
+
+		ctx, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond)
+		defer cancel()
+
+		callbackCalled := false
+		err = reader.readAll(ctx, "test-telemetry", func(context.Context, string, []byte) error {
+			callbackCalled = true
+			return errors.New("callback processing failed")
+		})
+
+		assert.Error(t, err) // Context timeout expected.
+		assert.True(t, callbackCalled, "Callback should have been called")
+
+		mockS3.AssertExpectations(t)
+		mockSQS.AssertExpectations(t)
+		// Verify DeleteMessage was never called - message should remain for retry.
+		mockSQS.AssertNotCalled(t, "DeleteMessage", mock.Anything, mock.Anything)
 	})
 }