From 452dd9197c44ee8238c1b0fcc54b670905f5372c Mon Sep 17 00:00:00 2001 From: Arve Knudsen Date: Thu, 25 Dec 2025 11:54:18 +0100 Subject: [PATCH] [receiver/awss3receiver] Fix data loss on partial SQS message processing failure The SQS notification reader was unconditionally deleting messages after processing all records, even when some S3 object retrievals or callback processing failed. This caused data loss when an SQS message contained multiple S3 notification records and any of them failed to process. Changes: - Track success/failure for all records in a message - Only delete message from SQS if ALL records processed successfully - Add log warning when message is left for retry due to failures - Add tests for partial failure, S3 retrieval error, and callback error The fix ensures messages remain in the queue for retry after visibility timeout when any record fails, preventing permanent data loss. Signed-off-by: Arve Knudsen --- ...awss3receiver-fix-sqs-partial-failure.yaml | 32 +++ receiver/awss3receiver/s3sqsreader.go | 26 ++- receiver/awss3receiver/s3sqsreader_test.go | 213 +++++++++++++++++- 3 files changed, 255 insertions(+), 16 deletions(-) create mode 100644 .chloggen/awss3receiver-fix-sqs-partial-failure.yaml diff --git a/.chloggen/awss3receiver-fix-sqs-partial-failure.yaml b/.chloggen/awss3receiver-fix-sqs-partial-failure.yaml new file mode 100644 index 0000000000000..996244c455109 --- /dev/null +++ b/.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] diff --git a/receiver/awss3receiver/s3sqsreader.go b/receiver/awss3receiver/s3sqsreader.go index bcd310eb898a8..b648bb8c313e1 100644 --- a/receiver/awss3receiver/s3sqsreader.go +++ b/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)) } } } diff --git a/receiver/awss3receiver/s3sqsreader_test.go b/receiver/awss3receiver/s3sqsreader_test.go index 1a4ac289b8b6e..86cefae7051a9 100644 --- a/receiver/awss3receiver/s3sqsreader_test.go +++ b/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) }) }