s3_input: handle s3:TestEvent messages to prevent redelivery

docs/modules/components/pages/inputs/aws_s3.adoc

@@ -107,6 +107,8 @@ If your notification events are being routed to SQS via an SNS topic then the ev
 
 When using SQS please make sure you have sensible values for `sqs.max_messages` and also the visibility timeout of the queue itself. When Redpanda Connect consumes an S3 object the SQS message that triggered it is not deleted until the S3 object has been sent onwards. This ensures at-least-once crash resiliency, but also means that if the S3 object takes longer to process than the visibility timeout of your queue then the same objects might be processed multiple times.
 
+Amazon S3 sends an `s3:TestEvent` notification whenever a bucket's event configuration is saved, to verify the queue is reachable. Redpanda Connect detects these (including via an SNS envelope) and deletes them automatically. Any other message with no extractable target key, for example due to a misconfigured `sqs.key_path`/`sqs.bucket_path`, is logged as a warning and left on the queue instead.
+
 == Download large files
 
 When downloading large files it's often necessary to process it in streamed parts in order to avoid loading the entire file in memory at a given time. In order to do this a <<scanner, `scanner`>> can be specified that determines how to break the input into smaller individual messages.

internal/impl/aws/s3/input.go

@@ -152,6 +152,8 @@ If your notification events are being routed to SQS via an SNS topic then the ev
 
 When using SQS please make sure you have sensible values for `+"`sqs.max_messages`"+` and also the visibility timeout of the queue itself. When Redpanda Connect consumes an S3 object the SQS message that triggered it is not deleted until the S3 object has been sent onwards. This ensures at-least-once crash resiliency, but also means that if the S3 object takes longer to process than the visibility timeout of your queue then the same objects might be processed multiple times.
 
+Amazon S3 sends an `+"`s3:TestEvent`"+` notification whenever a bucket's event configuration is saved, to verify the queue is reachable. Redpanda Connect detects these (including via an SNS envelope) and deletes them automatically. Any other message with no extractable target key, for example due to a misconfigured `+"`sqs.key_path`"+`/`+"`sqs.bucket_path`"+`, is logged as a warning and left on the queue instead.
+
 == Download large files
 
 When downloading large files it's often necessary to process it in streamed parts in order to avoid loading the entire file in memory at a given time. In order to do this a `+"<<scanner, `scanner`>>"+` can be specified that determines how to break the input into smaller individual messages.
@@ -472,20 +474,20 @@ func digStrsFromSlices(slice []any) []string {
 	return strs
 }
 
-func (s *sqsTargetReader) parseObjectPaths(sqsMsg *string) ([]s3ObjectTarget, error) {
+func (s *sqsTargetReader) parseObjectPaths(sqsMsg *string) (*gabs.Container, []s3ObjectTarget, error) {
 	gObj, err := gabs.ParseJSON([]byte(*sqsMsg))
 	if err != nil {
-		return nil, fmt.Errorf("parsing SQS message: %v", err)
+		return nil, nil, fmt.Errorf("parsing SQS message: %v", err)
 	}
 
 	if s.conf.SQS.EnvelopePath != "" {
 		d := gObj.Path(s.conf.SQS.EnvelopePath).Data()
 		if str, ok := d.(string); ok {
 			if gObj, err = gabs.ParseJSON([]byte(str)); err != nil {
-				return nil, fmt.Errorf("parsing enveloped message: %v", err)
+				return nil, nil, fmt.Errorf("parsing enveloped message: %v", err)
 			}
 		} else {
-			return nil, fmt.Errorf("expected string at envelope path, found %T", d)
+			return nil, nil, fmt.Errorf("expected string at envelope path, found %T", d)
 		}
 	}
 
@@ -510,22 +512,23 @@ func (s *sqsTargetReader) parseObjectPaths(sqsMsg *string) ([]s3ObjectTarget, er
 	objects := make([]s3ObjectTarget, 0, len(keys))
 	for i, key := range keys {
 		if key, err = url.QueryUnescape(key); err != nil {
-			return nil, fmt.Errorf("parsing key from SQS message: %v", err)
+			return nil, nil, fmt.Errorf("parsing key from SQS message: %v", err)
 		}
 		bucket := s.conf.Bucket
 		if len(buckets) > i {
 			bucket = buckets[i]
 		}
 		if bucket == "" {
-			return nil, errors.New("required bucket was not found in SQS message")
+			return nil, nil, errors.New("required bucket was not found in SQS message")
 		}
 		objects = append(objects, s3ObjectTarget{
 			key:    key,
 			bucket: bucket,
 		})
 	}
 
-	return objects, nil
+	// return gabs.Container to save reparsing
+	return gObj, objects, nil
 }
 
 func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget, error) {
@@ -570,15 +573,25 @@ func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget,
 			continue
 		}
 
-		objects, err := s.parseObjectPaths(sqsMsg.Body)
+		gObj, objects, err := s.parseObjectPaths(sqsMsg.Body)
 		if err != nil {
 			addDudFn(sqsMsg)
 			s.log.Errorf("SQS extract key error: %v", err)
 			continue
 		}
 		if len(objects) == 0 {
+			if isS3TestEvent(gObj) {
+				s.log.Debugf("Received S3 test event, deleting: %s", *sqsMsg.Body)
+				if err := s.ackSQSMessage(ctx, sqsMsg); err != nil {
+					s.log.Errorf("Failed to delete SQS test event message: %v", err)
+				}
+				continue
+			}
 			addDudFn(sqsMsg)
-			s.log.Debug("Extracted zero target keys from SQS message")
+			s.log.Warnf(
+				"Extracted zero target keys from SQS message using key_path %q (bucket_path %q) - this likely indicates a misconfigured key_path/bucket_path, or an unrecognised notification event type: %s",
+				s.conf.SQS.KeyPath, s.conf.SQS.BucketPath, *sqsMsg.Body,
+			)
 			continue
 		}
 
@@ -662,6 +675,14 @@ func (s *sqsTargetReader) ackSQSMessage(ctx context.Context, msg sqstypes.Messag
 	return err
 }
 
+func isS3TestEvent(gObj *gabs.Container) bool {
+	if gObj == nil {
+		return false
+	}
+	event, ok := gObj.Path("Event").Data().(string)
+	return ok && event == "s3:TestEvent"
+}
+
 //------------------------------------------------------------------------------
 
 // AmazonS3 is a benthos reader.Type implementation that reads messages from an

internal/impl/aws/s3/input_test.go

@@ -0,0 +1,93 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package s3
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseObjectPathsSNSEnvelope(t *testing.T) {
+	tests := []struct {
+		name            string
+		envelopePath    string
+		body            string
+		expectedKey     string
+		expectedBucket  string
+		expectTestEvent bool
+	}{
+		{
+			name:            "s3 test event",
+			body:            `{"Service":"Amazon S3","Event":"s3:TestEvent","Time":"2025-08-19T17:34:58.550Z","Bucket":"bucket-test","RequestId":"N99ABJ6Q","HostId":"+3DhJHKGDGBwqSTufMSS1UgAMIoRovmGa9vkZwWIb1="}`,
+			expectTestEvent: true,
+		},
+		{
+			name:           "regular object created notification",
+			body:           `{"Records":[{"eventName":"ObjectCreated:Put","s3":{"bucket":{"name":"bucket-test"},"object":{"key":"foo.txt"}}}]}`,
+			expectedKey:    "foo.txt",
+			expectedBucket: "bucket-test",
+		},
+		{
+			name: "Event field is not a string",
+			body: `{"Event":123}`,
+		},
+		{
+			name: "Event field missing entirely",
+			body: `{"Service":"Amazon S3"}`,
+		},
+		{
+			name:            "SNS-enveloped test event",
+			envelopePath:    "Message",
+			body:            `{"Type":"Notification","MessageId":"abc","TopicArn":"arn:aws:sns:eu-west-1:000000000000:topic","Message":"{\"Service\":\"Amazon S3\",\"Event\":\"s3:TestEvent\",\"Time\":\"2025-08-19T17:34:58.550Z\",\"Bucket\":\"bucket-test\",\"RequestId\":\"N99ABJ6Q\",\"HostId\":\"+3DhJHKGDGBwqSTufMSS1UgAMIoRovmGa9vkZwWIb1=\"}"}`,
+			expectTestEvent: true,
+		},
+		{
+			name:           "SNS-enveloped object created notification",
+			envelopePath:   "Message",
+			body:           `{"Type":"Notification","MessageId":"abc","TopicArn":"arn:aws:sns:eu-west-1:000000000000:topic","Message":"{\"Records\":[{\"eventName\":\"ObjectCreated:Put\",\"s3\":{\"bucket\":{\"name\":\"bucket-test\"},\"object\":{\"key\":\"foo.txt\"}}}]}"}`,
+			expectedKey:    "foo.txt",
+			expectedBucket: "bucket-test",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			reader := &sqsTargetReader{
+				conf: s3iConfig{
+					SQS: s3iSQSConfig{
+						EnvelopePath: test.envelopePath,
+						KeyPath:      "Records.*.s3.object.key",
+						BucketPath:   "Records.*.s3.bucket.name",
+					},
+				},
+			}
+
+			gObj, objects, err := reader.parseObjectPaths(&test.body)
+			require.NoError(t, err)
+
+			if test.expectedKey == "" {
+				assert.Empty(t, objects)
+			} else {
+				require.Len(t, objects, 1)
+				assert.Equal(t, test.expectedKey, objects[0].key)
+				assert.Equal(t, test.expectedBucket, objects[0].bucket)
+			}
+
+			assert.Equal(t, test.expectTestEvent, isS3TestEvent(gObj))
+		})
+	}
+}