From 6f2abd8871fd49ac62ecc461bc6b69d70ae0d87c Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 16:36:39 +0100 Subject: [PATCH 1/6] s3_input: handle s3:TestEvent messages to prevent redelivery --- internal/impl/aws/s3/input.go | 16 ++++ internal/impl/aws/s3/integration_test.go | 100 +++++++++++++++++++++++ 2 files changed, 116 insertions(+) diff --git a/internal/impl/aws/s3/input.go b/internal/impl/aws/s3/input.go index 594b0c8d8a..17323788fa 100644 --- a/internal/impl/aws/s3/input.go +++ b/internal/impl/aws/s3/input.go @@ -577,6 +577,13 @@ func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget, continue } if len(objects) == 0 { + if isS3TestEvent(sqsMsg.Body) { + 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") continue @@ -662,6 +669,15 @@ func (s *sqsTargetReader) ackSQSMessage(ctx context.Context, msg sqstypes.Messag return err } +func isS3TestEvent(sqsMsg *string) bool { + gObj, err := gabs.ParseJSON([]byte(*sqsMsg)) + if err != 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 diff --git a/internal/impl/aws/s3/integration_test.go b/internal/impl/aws/s3/integration_test.go index aa8b702307..0bcb38971a 100644 --- a/internal/impl/aws/s3/integration_test.go +++ b/internal/impl/aws/s3/integration_test.go @@ -25,6 +25,8 @@ import ( awsconfig "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/credentials" "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/sqs" + sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -95,6 +97,90 @@ input: ) }) + t.Run("via_sqs_test_event_deleted", func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + id := fmt.Sprintf("testevent%d", time.Now().UnixNano()) + require.NoError(t, awstest.CreateBucketQueue(ctx, lsPort, lsPort, id)) + + sqsClient := newLocalStackSQSClient(t, lsPort) + queueURL := fmt.Sprintf("http://localhost:%s/000000000000/queue-%s", lsPort, id) + + testEventBody := fmt.Sprintf( + `{"Service":"Amazon S3","Event":"s3:TestEvent","Time":"%s","Bucket":"bucket-%s","RequestId":"N99ABJ6Q","HostId":"+3DhJHKGDGBwqSTufMSS1UgAMIoRovmGa9vkZwWIb1="}`, + time.Now().UTC().Format(time.RFC3339), id, + ) + _, err := sqsClient.SendMessage(ctx, &sqs.SendMessageInput{ + QueueUrl: aws.String(queueURL), + MessageBody: aws.String(testEventBody), + }) + require.NoError(t, err) + + yaml := fmt.Sprintf(` +input: + aws_s3: + bucket: bucket-%s + endpoint: http://localhost:%s + force_path_style_urls: true + region: eu-west-1 + delete_objects: true + sqs: + url: %s + key_path: Records.*.s3.object.key + endpoint: http://localhost:%s + wait_time_seconds: 1 + credentials: + id: xxxxx + secret: xxxxx + token: xxxxx +`, id, lsPort, queueURL, lsPort) + + builder := service.NewStreamBuilder() + require.NoError(t, builder.SetYAML(yaml)) + + require.NoError(t, builder.AddConsumerFunc(func(_ context.Context, m *service.Message) error { + b, _ := m.AsBytes() + t.Fatalf("did not expect any message to be emitted for an s3:TestEvent, got: %s", b) + return nil + })) + + stream, err := builder.Build() + require.NoError(t, err) + + runErr := make(chan error, 1) + runCtx, runCancel := context.WithCancel(ctx) + defer runCancel() + go func() { runErr <- stream.Run(runCtx) }() + + // The test event should be deleted from the queue rather than left + // to be received again indefinitely. + assert.Eventually(t, func() bool { + out, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: aws.String(queueURL), + AttributeNames: []sqstypes.QueueAttributeName{ + sqstypes.QueueAttributeNameApproximateNumberOfMessages, + sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible, + }, + }) + if err != nil { + return false + } + return out.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessages)] == "0" && + out.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible)] == "0" + }, 15*time.Second, 500*time.Millisecond, "test event message should have been deleted from the queue") + + require.NoError(t, stream.StopWithin(10*time.Second)) + select { + case err := <-runErr: + if err != nil && !errors.Is(err, context.Canceled) { + require.NoError(t, err) + } + case <-time.After(10 * time.Second): + t.Fatal("stream did not exit after StopWithin returned") + } + }) + t.Run("via_sqs_lines", func(t *testing.T) { template := ` output: @@ -392,3 +478,17 @@ func newLocalStackS3Client(t *testing.T, port string) *s3.Client { o.UsePathStyle = true }) } + +// newLocalStackSQSClient builds an SQS client pointed at the LocalStack +// instance on the supplied port, with dummy credentials. +func newLocalStackSQSClient(t *testing.T, port string) *sqs.Client { + t.Helper() + endpoint := fmt.Sprintf("http://localhost:%s", port) + cfg, err := awsconfig.LoadDefaultConfig(t.Context(), + awsconfig.WithRegion("eu-west-1"), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("xxxxx", "xxxxx", "xxxxx")), + ) + require.NoError(t, err) + cfg.BaseEndpoint = &endpoint + return sqs.NewFromConfig(cfg) +} From f5b49ef67ad72759c4501e5cdf821d4391c90741 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 16:59:36 +0100 Subject: [PATCH 2/6] input_s3: update integration test --- internal/impl/aws/s3/integration_test.go | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/internal/impl/aws/s3/integration_test.go b/internal/impl/aws/s3/integration_test.go index 0bcb38971a..2b7ec029a2 100644 --- a/internal/impl/aws/s3/integration_test.go +++ b/internal/impl/aws/s3/integration_test.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "sync" "testing" "time" @@ -139,9 +140,23 @@ input: builder := service.NewStreamBuilder() require.NoError(t, builder.SetYAML(yaml)) + // The consumer func runs on the stream's own goroutine, not the test + // goroutine, so failures here must not call t.Fatalf/require (which + // call FailNow, only valid from the test goroutine). Instead, record + // any unexpected message under a mutex and assert on it below, once + // the stream has stopped. + var ( + unexpectedMu sync.Mutex + unexpectedMsgs [][]byte + ) require.NoError(t, builder.AddConsumerFunc(func(_ context.Context, m *service.Message) error { - b, _ := m.AsBytes() - t.Fatalf("did not expect any message to be emitted for an s3:TestEvent, got: %s", b) + b, err := m.AsBytes() + if err != nil { + b = fmt.Appendf(nil, "", err) + } + unexpectedMu.Lock() + unexpectedMsgs = append(unexpectedMsgs, b) + unexpectedMu.Unlock() return nil })) @@ -179,6 +194,10 @@ input: case <-time.After(10 * time.Second): t.Fatal("stream did not exit after StopWithin returned") } + + unexpectedMu.Lock() + assert.Empty(t, unexpectedMsgs, "did not expect any message to be emitted for an s3:TestEvent") + unexpectedMu.Unlock() }) t.Run("via_sqs_lines", func(t *testing.T) { From bd7c8d71e396ef9b17de609cb15649d1dcefa62b Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 17:28:15 +0100 Subject: [PATCH 3/6] input_s3: result parsed object --- internal/impl/aws/s3/input.go | 24 ++++---- internal/impl/aws/s3/input_test.go | 93 ++++++++++++++++++++++++++++++ 2 files changed, 105 insertions(+), 12 deletions(-) create mode 100644 internal/impl/aws/s3/input_test.go diff --git a/internal/impl/aws/s3/input.go b/internal/impl/aws/s3/input.go index 17323788fa..30066123a6 100644 --- a/internal/impl/aws/s3/input.go +++ b/internal/impl/aws/s3/input.go @@ -472,20 +472,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,14 +510,14 @@ 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, @@ -525,7 +525,8 @@ func (s *sqsTargetReader) parseObjectPaths(sqsMsg *string) ([]s3ObjectTarget, er }) } - return objects, nil + // return gabs.Container to save reparsing + return gObj, objects, nil } func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget, error) { @@ -570,14 +571,14 @@ 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(sqsMsg.Body) { + 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) @@ -669,9 +670,8 @@ func (s *sqsTargetReader) ackSQSMessage(ctx context.Context, msg sqstypes.Messag return err } -func isS3TestEvent(sqsMsg *string) bool { - gObj, err := gabs.ParseJSON([]byte(*sqsMsg)) - if err != nil { +func isS3TestEvent(gObj *gabs.Container) bool { + if gObj == nil { return false } event, ok := gObj.Path("Event").Data().(string) diff --git a/internal/impl/aws/s3/input_test.go b/internal/impl/aws/s3/input_test.go new file mode 100644 index 0000000000..319081cc01 --- /dev/null +++ b/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)) + }) + } +} From 8a125a5cb7173c84aa8752c252a479c54213e57d Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 17:33:46 +0100 Subject: [PATCH 4/6] input_s3: remove redundant comment --- internal/impl/aws/s3/integration_test.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/internal/impl/aws/s3/integration_test.go b/internal/impl/aws/s3/integration_test.go index 2b7ec029a2..273d41e758 100644 --- a/internal/impl/aws/s3/integration_test.go +++ b/internal/impl/aws/s3/integration_test.go @@ -140,11 +140,6 @@ input: builder := service.NewStreamBuilder() require.NoError(t, builder.SetYAML(yaml)) - // The consumer func runs on the stream's own goroutine, not the test - // goroutine, so failures here must not call t.Fatalf/require (which - // call FailNow, only valid from the test goroutine). Instead, record - // any unexpected message under a mutex and assert on it below, once - // the stream has stopped. var ( unexpectedMu sync.Mutex unexpectedMsgs [][]byte From 10b4a2efdc9cb77a0d08ed004db95f37287a45ec Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 17:58:07 +0100 Subject: [PATCH 5/6] input_s3: log zero-key messages that are not test events as warn --- internal/impl/aws/s3/input.go | 5 +- internal/impl/aws/s3/integration_test.go | 113 +++++++++++++++++++++++ 2 files changed, 117 insertions(+), 1 deletion(-) diff --git a/internal/impl/aws/s3/input.go b/internal/impl/aws/s3/input.go index 30066123a6..bb44650e50 100644 --- a/internal/impl/aws/s3/input.go +++ b/internal/impl/aws/s3/input.go @@ -586,7 +586,10 @@ func (s *sqsTargetReader) readSQSEvents(ctx context.Context) ([]*s3ObjectTarget, 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 } diff --git a/internal/impl/aws/s3/integration_test.go b/internal/impl/aws/s3/integration_test.go index 273d41e758..dea660dbcb 100644 --- a/internal/impl/aws/s3/integration_test.go +++ b/internal/impl/aws/s3/integration_test.go @@ -15,9 +15,12 @@ package s3 import ( + "bytes" "context" "errors" "fmt" + "log/slog" + "strings" "sync" "testing" "time" @@ -195,6 +198,99 @@ input: unexpectedMu.Unlock() }) + t.Run("via_sqs_zero_key_non_test_event_warns", func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + id := fmt.Sprintf("zerokey%d", time.Now().UnixNano()) + require.NoError(t, awstest.CreateBucketQueue(ctx, lsPort, lsPort, id)) + + sqsClient := newLocalStackSQSClient(t, lsPort) + queueURL := fmt.Sprintf("http://localhost:%s/000000000000/queue-%s", lsPort, id) + + // Valid JSON, not an s3:TestEvent, but missing the object key at the + // configured key_path - simulates a key_path/bucket_path + // misconfiguration (or an unrecognised event type) rather than a + // test event, which should be handled differently: warned about and + // left on the queue, not silently deleted. + misconfiguredBody := fmt.Sprintf( + `{"Records":[{"eventName":"ObjectCreated:Put","s3":{"bucket":{"name":"bucket-%s"}}}]}`, id, + ) + _, err := sqsClient.SendMessage(ctx, &sqs.SendMessageInput{ + QueueUrl: aws.String(queueURL), + MessageBody: aws.String(misconfiguredBody), + }) + require.NoError(t, err) + + yaml := fmt.Sprintf(` +input: + aws_s3: + bucket: bucket-%s + endpoint: http://localhost:%s + force_path_style_urls: true + region: eu-west-1 + delete_objects: true + sqs: + url: %s + key_path: Records.*.s3.object.key + endpoint: http://localhost:%s + wait_time_seconds: 1 + credentials: + id: xxxxx + secret: xxxxx + token: xxxxx +`, id, lsPort, queueURL, lsPort) + + var logBuf syncBuffer + builder := service.NewStreamBuilder() + builder.SetLogger(slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelDebug}))) + + require.NoError(t, builder.SetYAML(yaml)) + require.NoError(t, builder.AddConsumerFunc(func(_ context.Context, m *service.Message) error { + b, _ := m.AsBytes() + t.Errorf("did not expect any message to be emitted for a zero-key notification, got: %s", b) + return nil + })) + + stream, err := builder.Build() + require.NoError(t, err) + + runErr := make(chan error, 1) + runCtx, runCancel := context.WithCancel(ctx) + defer runCancel() + go func() { runErr <- stream.Run(runCtx) }() + + // A non-test-event, zero-key message should be warned about, + // identifying the likely key_path misconfiguration, and - unlike an + // s3:TestEvent - left on the queue for redelivery rather than + // deleted. + assert.Eventually(t, func() bool { + return strings.Contains(logBuf.String(), "level=WARN") && strings.Contains(logBuf.String(), "key_path") + }, 15*time.Second, 500*time.Millisecond, "expected a WARN log identifying a key_path misconfiguration, got logs: %s", &logBuf) + + out, err := sqsClient.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: aws.String(queueURL), + AttributeNames: []sqstypes.QueueAttributeName{ + sqstypes.QueueAttributeNameApproximateNumberOfMessages, + sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible, + }, + }) + require.NoError(t, err) + visible := out.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessages)] + notVisible := out.Attributes[string(sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible)] + assert.False(t, visible == "0" && notVisible == "0", "misconfigured message should remain on the queue for redelivery, not be deleted") + + require.NoError(t, stream.StopWithin(10*time.Second)) + select { + case err := <-runErr: + if err != nil && !errors.Is(err, context.Canceled) { + require.NoError(t, err) + } + case <-time.After(10 * time.Second): + t.Fatal("stream did not exit after StopWithin returned") + } + }) + t.Run("via_sqs_lines", func(t *testing.T) { template := ` output: @@ -506,3 +602,20 @@ func newLocalStackSQSClient(t *testing.T, port string) *sqs.Client { cfg.BaseEndpoint = &endpoint return sqs.NewFromConfig(cfg) } + +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} From c8b5732d3b526cc9a2c255d40098eb2cfa0836c4 Mon Sep 17 00:00:00 2001 From: Joseph Woodward Date: Tue, 4 Aug 2026 18:21:30 +0100 Subject: [PATCH 6/6] input_s3: document new behaviour --- docs/modules/components/pages/inputs/aws_s3.adoc | 2 ++ internal/impl/aws/s3/input.go | 2 ++ 2 files changed, 4 insertions(+) diff --git a/docs/modules/components/pages/inputs/aws_s3.adoc b/docs/modules/components/pages/inputs/aws_s3.adoc index 9c9a62a6c8..cabb3d3154 100644 --- a/docs/modules/components/pages/inputs/aws_s3.adoc +++ b/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 <> can be specified that determines how to break the input into smaller individual messages. diff --git a/internal/impl/aws/s3/input.go b/internal/impl/aws/s3/input.go index bb44650e50..2159ddc126 100644 --- a/internal/impl/aws/s3/input.go +++ b/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 `+"<>"+` can be specified that determines how to break the input into smaller individual messages.