diff --git a/.ci/jobs/docker-compose.yml b/.ci/jobs/docker-compose.yml
new file mode 100644
index 000000000000..e9fc43ff7046
--- /dev/null
+++ b/.ci/jobs/docker-compose.yml
@@ -0,0 +1,23 @@
+version: '2.3'
+services:
+  # This is a proxy used to block beats until all services are healthy.
+  # See: https://github.com/docker/compose/issues/4369
+  proxy_dep:
+    image: busybox
+    depends_on:
+      localstack: { condition: service_healthy }
+      
+  localstack:
+    container_name: "${localstack_integration_test_container}"
+    image: localstack/localstack:2.1.0 # Latest stable release
+    ports:
+      - "127.0.0.1:4566:4566"            # LocalStack Gateway
+    environment:
+      - DEBUG=1
+      - DOCKER_HOST=unix:///var/run/docker.sock
+      - LOCALSTACK_HOST=localhost
+      - S3_HOSTNAME=localhost
+      - PROVIDER_OVERRIDE_S3=asf
+    volumes:
+      - "${LOCALSTACK_VOLUME_DIR:-./volume}:/var/lib/localstack"
+      - "/var/run/docker.sock:/var/run/docker.sock"
diff --git a/.ci/scripts/docker-services-cleanup.sh b/.ci/scripts/docker-services-cleanup.sh
new file mode 100755
index 000000000000..cc182413a2e6
--- /dev/null
+++ b/.ci/scripts/docker-services-cleanup.sh
@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+
+set -exuo pipefail
+
+${HOME}/bin/docker-compose -f .ci/jobs/docker-compose.yml down -v
+
+exit $?
diff --git a/.ci/scripts/install-docker-services.sh b/.ci/scripts/install-docker-services.sh
new file mode 100755
index 000000000000..420362f83557
--- /dev/null
+++ b/.ci/scripts/install-docker-services.sh
@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+
+set -exuo pipefail
+
+${HOME}/bin/docker-compose -f .ci/jobs/docker-compose.yml up -d
+
+exit $?
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 49595dea612b..59d25cec04a2 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -356,6 +356,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Add `clean_session` configuration setting for MQTT input.  {pull}35806[16204]
 - Add fingerprint mode for the filestream scanner and new file identity based on it {issue}34419[34419] {pull}35734[35734]
 - Add file system metadata to events ingested via filestream {issue}35801[35801] {pull}36065[36065]
+- Add support for localstack based input integration testing {pull}35727[35727]
 - Allow parsing bytes in and bytes out as long integer in CEF processor. {issue}36100[36100] {pull}36108[36108]
 - Add support for registered owners and users to AzureAD entity analytics provider. {pull}36092[36092]
 
diff --git a/Jenkinsfile b/Jenkinsfile
index cdee0f662cdb..a0231b94957c 100644
--- a/Jenkinsfile
+++ b/Jenkinsfile
@@ -618,7 +618,7 @@ def targetWithoutNode(Map args = [:]) {
               cmd(label: "${args.id?.trim() ? args.id : env.STAGE_NAME} - ${command}", script: "${command}")
             }
           } else {
-            cmd(label: "${args.id?.trim() ? args.id : env.STAGE_NAME} - ${command}", script: "${command}")
+              cmd(label: "${args.id?.trim() ? args.id : env.STAGE_NAME} - ${command}", script: "${command}")
           }
         }
       }
@@ -934,6 +934,8 @@ def startCloudTestEnv(Map args = [:]) {
   stage("${name}-prepare-cloud-env"){
     withBeatsEnv(archive: false, withModule: false) {
       try {
+        // Run the docker services to setup the emulated cloud environment
+        sh(label: 'Run docker-compose services for emulated cloud env', script: ".ci/scripts/install-docker-services.sh ", returnStatus: true)
         dirs?.each { folder ->
           retryWithSleep(retries: 2, seconds: 5, backoff: true){
             terraformApply(folder)
@@ -944,9 +946,12 @@ def startCloudTestEnv(Map args = [:]) {
           // If it failed then cleanup without failing the build
           sh(label: 'Terraform Cleanup', script: ".ci/scripts/terraform-cleanup.sh ${folder}", returnStatus: true)
         }
+        // Cleanup the docker services
+        sh(label: 'Docker Compose Cleanup', script: ".ci/scripts/docker-services-cleanup.sh", returnStatus: true)
+        
         error('startCloudTestEnv: terraform apply failed.')
       } finally {
-        dirs?.each { folder ->
+          dirs?.each { folder ->
           // Archive terraform states in case manual cleanup is needed.
           archiveArtifacts(allowEmptyArchive: true, artifacts: '**/terraform.tfstate')
           dir("${folder}") {
@@ -978,6 +983,7 @@ def terraformApply(String directory) {
 * Tear down the terraform environments, by looking for all terraform states in directory
 * then it runs terraform destroy for each one.
 * It uses terraform states previously stashed by startCloudTestEnv.
+* This also tears down any associated docker services
 */
 def terraformCleanup(Map args = [:]) {
   String name = normalise(args.name)
@@ -988,6 +994,8 @@ def terraformCleanup(Map args = [:]) {
       retryWithSleep(retries: 2, seconds: 5, backoff: true) {
         sh(label: "Terraform Cleanup", script: ".ci/scripts/terraform-cleanup.sh ${directory}")
       }
+      // Cleanup associated docker services
+      sh(label: 'Docker Compose Cleanup', script: ".ci/scripts/docker-services-cleanup.sh")
     }
   }
 }
diff --git a/x-pack/filebeat/input/awss3/_meta/terraform/.gitignore b/x-pack/filebeat/input/awss3/_meta/terraform/.gitignore
index 0825744a7760..1af7b09a151d 100644
--- a/x-pack/filebeat/input/awss3/_meta/terraform/.gitignore
+++ b/x-pack/filebeat/input/awss3/_meta/terraform/.gitignore
@@ -1,3 +1,3 @@
 terraform/
-outputs.yml
+outputs*.yml
 *.tfstate*
diff --git a/x-pack/filebeat/input/awss3/_meta/terraform/README.md b/x-pack/filebeat/input/awss3/_meta/terraform/README.md
index d5614b99a92e..41100d98dad4 100644
--- a/x-pack/filebeat/input/awss3/_meta/terraform/README.md
+++ b/x-pack/filebeat/input/awss3/_meta/terraform/README.md
@@ -5,6 +5,8 @@ for executing the integration tests for the `aws-s3` Filebeat input. It creates
 an S3 bucket and SQS queue and configures S3 `ObjectCreated:*` notifications to
 be delivered to SQS. It also creates a second S3 bucket, SNS topic, SQS queue and configures S3 `ObjectCreated:*` notifications to be delivered to SNS and also creates a subscription for this SNS topic to SQS queue to automatically place messages sent to SNS topic in SQS queue.
 
+## Cloud AWS environment
+
 It outputs configuration information that is consumed by the tests to
 `outputs.yml`. The AWS resources are randomly named to prevent name collisions
 between multiple users.
@@ -42,4 +44,40 @@ the S3 bucket and its contents.
 
     `terraform destroy`
 
+## Emulated cloud Localstack environment
+
+It outputs configuration information that is consumed by the tests to
+`outputs-localstack.yml`. The AWS resources are randomly named to prevent name collisions
+between multiple users.
+
+### Usage
+
+You must have the appropriate Localstack environment up and running in docker. 
+You can use `.ci/jobs/docker-compose.yml` to spin up localstack environment.
+
+1. Execute terraform in this directory to create the resources. This will also
+write the `outputs-localstack.yml`. You can use `export TF_VAR_aws_region=NNNNN` in order
+to match the AWS region of the profile you are using.
+
+    `terraform apply`
+
+
+2. (Optional) View the output configuration.
+
+   ```yaml
+   "aws_region": "us-east-1"
+   "bucket_name": "filebeat-s3-integtest-8iok1h"
+   "queue_url": "https://localhost:4566/000000000000/filebeat-s3-integtest-8iok1h"
+   ```
+
+4. Execute the integration test.
+
+    ```
+    cd x-pack/filebeat/input/awss3
+    go test -tags aws,integration -run TestInputRun*Localstack* -v .
+    ```
+
+5. Cleanup AWS resources. Execute terraform to remove the SQS queue and delete
+the S3 bucket and its contents.
 
+    `terraform destroy`
\ No newline at end of file
diff --git a/x-pack/filebeat/input/awss3/_meta/terraform/localstack.tf b/x-pack/filebeat/input/awss3/_meta/terraform/localstack.tf
new file mode 100644
index 000000000000..2d2fcbe09ecf
--- /dev/null
+++ b/x-pack/filebeat/input/awss3/_meta/terraform/localstack.tf
@@ -0,0 +1,89 @@
+provider "aws" {
+  alias = "localstack"
+  access_key                  = "bharat"
+  secret_key                  = "bharat"
+  region                      = "us-east-1"
+  s3_use_path_style           = true
+  skip_credentials_validation = true
+  skip_metadata_api_check     = true
+  skip_requesting_account_id  = true
+
+  endpoints {
+    apigateway     = "http://localhost:4566"
+    apigatewayv2   = "http://localhost:4566"
+    cloudformation = "http://localhost:4566"
+    cloudwatch     = "http://localhost:4566"
+    dynamodb       = "http://localhost:4566"
+    ec2            = "http://localhost:4566"
+    es             = "http://localhost:4566"
+    elasticache    = "http://localhost:4566"
+    firehose       = "http://localhost:4566"
+    iam            = "http://localhost:4566"
+    kinesis        = "http://localhost:4566"
+    lambda         = "http://localhost:4566"
+    rds            = "http://localhost:4566"
+    redshift       = "http://localhost:4566"
+    route53        = "http://localhost:4566"
+    s3             = "http://localhost:4566"
+    secretsmanager = "http://localhost:4566"
+    ses            = "http://localhost:4566"
+    sns            = "http://localhost:4566"
+    sqs            = "http://localhost:4566"
+    ssm            = "http://localhost:4566"
+    stepfunctions  = "http://localhost:4566"
+    sts            = "http://localhost:4566"
+  }
+}
+
+resource "random_string" "random_localstack" {
+  length  = 6
+  special = false
+  upper   = false
+}
+
+resource "aws_s3_bucket" "filebeat-integtest-localstack" {
+  provider = aws.localstack
+  bucket        = "filebeat-s3-integtest-localstack-${random_string.random_localstack.result}"
+  force_destroy = true
+}
+
+resource "aws_sqs_queue" "filebeat-integtest-localstack" {
+  provider = aws.localstack
+  name   = "filebeat-sqs-integtest-localstack-${random_string.random_localstack.result}"
+  policy = <<POLICY
+{
+  "Version": "2012-10-17",
+  "Id": "sqspolicy",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": "*",
+      "Action": "sqs:SendMessage",
+      "Resource": "arn:aws:sqs:*:*:filebeat-sqs-integtest-localstack-${random_string.random_localstack.result}",
+      "Condition": {
+        "ArnEquals": { "aws:SourceArn": "${aws_s3_bucket.filebeat-integtest-localstack.arn}" }
+      }
+    }
+  ]
+}
+POLICY
+
+  depends_on = [
+    aws_s3_bucket.filebeat-integtest-localstack,
+  ]
+}
+
+resource "aws_s3_bucket_notification" "bucket_notification-localstack" {
+  provider = aws.localstack
+  bucket = aws_s3_bucket.filebeat-integtest-localstack.id
+
+  queue {
+    queue_arn = aws_sqs_queue.filebeat-integtest-localstack.arn
+    events    = ["s3:ObjectCreated:*"]
+  }
+
+  depends_on = [
+    aws_s3_bucket.filebeat-integtest-localstack,
+    aws_sqs_queue.filebeat-integtest-localstack,
+  ]
+}
diff --git a/x-pack/filebeat/input/awss3/_meta/terraform/outputs.tf b/x-pack/filebeat/input/awss3/_meta/terraform/outputs.tf
index a9851024bb05..e95983a72378 100644
--- a/x-pack/filebeat/input/awss3/_meta/terraform/outputs.tf
+++ b/x-pack/filebeat/input/awss3/_meta/terraform/outputs.tf
@@ -9,3 +9,13 @@ resource "local_file" "secrets" {
   filename        = "${path.module}/outputs.yml"
   file_permission = "0644"
 }
+
+resource "local_file" "secrets-localstack" {
+  content = yamlencode({
+    "queue_url" : aws_sqs_queue.filebeat-integtest-localstack.url
+    "aws_region" : aws_s3_bucket.filebeat-integtest-localstack.region
+    "bucket_name" : aws_s3_bucket.filebeat-integtest-localstack.id
+  })
+  filename        = "${path.module}/outputs-localstack.yml"
+  file_permission = "0644"
+}
diff --git a/x-pack/filebeat/input/awss3/input.go b/x-pack/filebeat/input/awss3/input.go
index fa2bedfeebe7..c50b4bb4c42d 100644
--- a/x-pack/filebeat/input/awss3/input.go
+++ b/x-pack/filebeat/input/awss3/input.go
@@ -190,6 +190,7 @@ func (in *s3Input) createSQSReceiver(ctx v2.Context, pipeline beat.Pipeline) (*s
 			if in.config.AWSConfig.FIPSEnabled {
 				o.EndpointOptions.UseFIPSEndpoint = awssdk.FIPSEndpointStateEnabled
 			}
+			o.UsePathStyle = in.config.PathStyle
 		}),
 	}
 
diff --git a/x-pack/filebeat/input/awss3/input_integration_test.go b/x-pack/filebeat/input/awss3/input_integration_test.go
index c843e47e93f6..62cbc8350113 100644
--- a/x-pack/filebeat/input/awss3/input_integration_test.go
+++ b/x-pack/filebeat/input/awss3/input_integration_test.go
@@ -40,7 +40,8 @@ const (
 )
 
 const (
-	terraformOutputYML = "_meta/terraform/outputs.yml"
+	terraformOutputYML   = "_meta/terraform/outputs.yml"
+	terraformOutputLsYML = "_meta/terraform/outputs-localstack.yml"
 )
 
 type terraformOutputData struct {
@@ -51,13 +52,21 @@ type terraformOutputData struct {
 	QueueURLForSNS   string `yaml:"queue_url_for_sns"`
 }
 
-func getTerraformOutputs(t *testing.T) terraformOutputData {
+func getTerraformOutputs(t *testing.T, isLocalStack bool) terraformOutputData {
 	t.Helper()
 
 	_, filename, _, _ := runtime.Caller(0)
-	ymlData, err := ioutil.ReadFile(path.Join(path.Dir(filename), terraformOutputYML))
+
+	var outputFile string
+	if isLocalStack {
+		outputFile = terraformOutputLsYML
+	} else {
+		outputFile = terraformOutputYML
+	}
+
+	ymlData, err := ioutil.ReadFile(path.Join(path.Dir(filename), outputFile))
 	if os.IsNotExist(err) {
-		t.Skipf("Run 'terraform apply' in %v to setup S3 and SQS for the test.", filepath.Dir(terraformOutputYML))
+		t.Skipf("Run 'terraform apply' in %v to setup S3 and SQS for the test.", filepath.Dir(outputFile))
 	}
 	if err != nil {
 		t.Fatalf("failed reading terraform output data: %v", err)
@@ -65,7 +74,6 @@ func getTerraformOutputs(t *testing.T) terraformOutputData {
 
 	var rtn terraformOutputData
 	dec := yaml.NewDecoder(bytes.NewReader(ymlData))
-	dec.SetStrict(true)
 	if err = dec.Decode(&rtn); err != nil {
 		t.Fatal(err)
 	}
@@ -145,16 +153,127 @@ func newV2Context() (v2.Context, func()) {
 	}, cancel
 }
 
+// Creates a default config for Localstack based tests
+func defaultTestConfig(region, queueURL string) config {
+	c := config{
+		APITimeout:          120 * time.Second,
+		VisibilityTimeout:   300 * time.Second,
+		BucketListInterval:  120 * time.Second,
+		BucketListPrefix:    "",
+		SQSWaitTime:         20 * time.Second,
+		SQSMaxReceiveCount:  5,
+		MaxNumberOfMessages: 5,
+		PathStyle:           true,
+		RegionName:          region,
+		QueueURL:            queueURL,
+	}
+	c.ReaderConfig.InitDefaults()
+	return c
+}
+
+// Create an aws config for Localstack based tests
+func makeLocalstackConfig(awsRegion string) (aws.Config, error) {
+	awsLocalstackEndpoint := "http://localhost:4566" // Default Localstack endpoint
+
+	// Add a custom endpointResolver to the awsConfig so that all the requests are routed to this endpoint
+	customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
+		return aws.Endpoint{
+			PartitionID:   "aws",
+			URL:           awsLocalstackEndpoint,
+			SigningRegion: awsRegion,
+		}, nil
+	})
+
+	return awsConfig.LoadDefaultConfig(context.TODO(),
+		awsConfig.WithRegion(awsRegion),
+		awsConfig.WithEndpointResolverWithOptions(customResolver),
+	)
+}
+
+// Tests reading SQS notifcation via awss3 input when an object is PUT in S3
+// and a notification is generated to SQS on Localstack
+func TestInputRunSQSOnLocalstack(t *testing.T) {
+	logp.TestingSetup()
+
+	// Terraform is used to set up S3,SQS and must be executed manually.
+	tfConfig := getTerraformOutputs(t, true)
+
+	// Read the necessary terraform outputs
+	region := tfConfig.AWSRegion
+	bucketName := tfConfig.BucketName
+	queueUrl := tfConfig.QueueURL
+
+	// Create a default config for the awss3 input
+	config := defaultTestConfig(region, queueUrl)
+	awsCfg, err := makeLocalstackConfig(region)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Ensure SQS is empty before testing.
+	drainSQS(t, region, queueUrl, awsCfg)
+
+	// Upload test files to S3 to generate SQS notification
+	s3Client := s3.NewFromConfig(awsCfg, func(o *s3.Options) {
+		o.UsePathStyle = true
+	})
+	uploadS3TestFiles(t, region, bucketName, s3Client,
+		"testdata/events-array.json",
+		"testdata/invalid.json",
+		"testdata/log.json",
+		"testdata/log.ndjson",
+		"testdata/multiline.json",
+		"testdata/multiline.json.gz",
+		"testdata/multiline.txt",
+		"testdata/log.txt",
+	)
+
+	inputCtx, cancel := newV2Context()
+	t.Cleanup(cancel)
+	time.AfterFunc(15*time.Second, func() {
+		cancel()
+	})
+
+	// Initialize s3Input with the test config
+	s3Input := &s3Input{
+		config:    config,
+		awsConfig: awsCfg,
+		store:     openTestStatestore(),
+	}
+	// Run S3 Input with desired context
+	var errGroup errgroup.Group
+	errGroup.Go(func() error {
+		return s3Input.Run(inputCtx, &fakePipeline{})
+	})
+
+	if err := errGroup.Wait(); err != nil {
+		t.Fatal(err)
+	}
+
+	assert.EqualValues(t, s3Input.metrics.sqsMessagesReceivedTotal.Get(), 8) // S3 could batch notifications.
+	assert.EqualValues(t, s3Input.metrics.sqsMessagesInflight.Get(), 0)
+	assert.EqualValues(t, s3Input.metrics.sqsMessagesDeletedTotal.Get(), 7)
+	assert.EqualValues(t, s3Input.metrics.sqsMessagesReturnedTotal.Get(), 1) // Invalid JSON is returned so that it can eventually be DLQed.
+	assert.EqualValues(t, s3Input.metrics.sqsVisibilityTimeoutExtensionsTotal.Get(), 0)
+	assert.EqualValues(t, s3Input.metrics.s3ObjectsInflight.Get(), 0)
+	assert.EqualValues(t, s3Input.metrics.s3ObjectsRequestedTotal.Get(), 8)
+	assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), uint64(0x13))
+	assert.Greater(t, s3Input.metrics.sqsLagTime.Mean(), 0.0)
+	assert.EqualValues(t, s3Input.metrics.sqsWorkerUtilization.Get(), 0.0) // Workers are reset after processing and hence utilization should be 0 at the end
+}
+
 func TestInputRunSQS(t *testing.T) {
 	logp.TestingSetup()
 
 	// Terraform is used to set up S3 and SQS and must be executed manually.
-	tfConfig := getTerraformOutputs(t)
+	tfConfig := getTerraformOutputs(t, false)
+	awsCfg := makeAWSConfig(t, tfConfig.AWSRegion)
 
 	// Ensure SQS is empty before testing.
-	drainSQS(t, tfConfig.AWSRegion, tfConfig.QueueURL)
+	drainSQS(t, tfConfig.AWSRegion, tfConfig.QueueURL, awsCfg)
 
-	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName,
+	s3Client := s3.NewFromConfig(awsCfg)
+	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName, s3Client,
 		"testdata/events-array.json",
 		"testdata/invalid.json",
 		"testdata/log.json",
@@ -198,9 +317,11 @@ func TestInputRunS3(t *testing.T) {
 	logp.TestingSetup()
 
 	// Terraform is used to set up S3 and must be executed manually.
-	tfConfig := getTerraformOutputs(t)
+	tfConfig := getTerraformOutputs(t, false)
+	awsCfg := makeAWSConfig(t, tfConfig.AWSRegion)
 
-	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName,
+	s3Client := s3.NewFromConfig(awsCfg)
+	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName, s3Client,
 		"testdata/events-array.json",
 		"testdata/invalid.json",
 		"testdata/log.json",
@@ -236,15 +357,9 @@ func TestInputRunS3(t *testing.T) {
 	assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), 12)
 }
 
-func uploadS3TestFiles(t *testing.T, region, bucket string, filenames ...string) {
+func uploadS3TestFiles(t *testing.T, region, bucket string, s3Client *s3.Client, filenames ...string) {
 	t.Helper()
 
-	cfg, err := awsConfig.LoadDefaultConfig(context.TODO())
-	if err != nil {
-		t.Fatal(err)
-	}
-	cfg.Region = region
-	s3Client := s3.NewFromConfig(cfg)
 	uploader := s3manager.NewUploader(s3Client)
 
 	_, basefile, _, _ := runtime.Caller(0)
@@ -276,13 +391,16 @@ func uploadS3TestFiles(t *testing.T, region, bucket string, filenames ...string)
 	}
 }
 
-func drainSQS(t *testing.T, region string, queueURL string) {
+func makeAWSConfig(t *testing.T, region string) aws.Config {
 	cfg, err := awsConfig.LoadDefaultConfig(context.TODO())
 	if err != nil {
 		t.Fatal(err)
 	}
 	cfg.Region = region
+	return cfg
+}
 
+func drainSQS(t *testing.T, region string, queueURL string, cfg aws.Config) {
 	sqs := &awsSQSAPI{
 		client:            sqs.NewFromConfig(cfg),
 		queueURL:          queueURL,
@@ -321,7 +439,7 @@ func TestGetRegionForBucketARN(t *testing.T) {
 	logp.TestingSetup()
 
 	// Terraform is used to set up S3 and must be executed manually.
-	tfConfig := getTerraformOutputs(t)
+	tfConfig := getTerraformOutputs(t, false)
 
 	cfg, err := awsConfig.LoadDefaultConfig(context.TODO())
 	if err != nil {
@@ -339,9 +457,11 @@ func TestPaginatorListPrefix(t *testing.T) {
 	logp.TestingSetup()
 
 	// Terraform is used to set up S3 and must be executed manually.
-	tfConfig := getTerraformOutputs(t)
+	tfConfig := getTerraformOutputs(t, false)
+	awsCfg := makeAWSConfig(t, tfConfig.AWSRegion)
 
-	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName,
+	s3Client := s3.NewFromConfig(awsCfg)
+	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName, s3Client,
 		"testdata/events-array.json",
 		"testdata/invalid.json",
 		"testdata/log.json",
@@ -358,8 +478,6 @@ func TestPaginatorListPrefix(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	s3Client := s3.NewFromConfig(cfg)
-
 	s3API := &awsS3API{
 		client: s3Client,
 	}
@@ -387,12 +505,14 @@ func TestInputRunSNS(t *testing.T) {
 	logp.TestingSetup()
 
 	// Terraform is used to set up S3, SNS and SQS and must be executed manually.
-	tfConfig := getTerraformOutputs(t)
+	tfConfig := getTerraformOutputs(t, false)
+	awsCfg := makeAWSConfig(t, tfConfig.AWSRegion)
 
 	// Ensure SQS is empty before testing.
-	drainSQS(t, tfConfig.AWSRegion, tfConfig.QueueURLForSNS)
+	drainSQS(t, tfConfig.AWSRegion, tfConfig.QueueURLForSNS, awsCfg)
 
-	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketNameForSNS,
+	s3Client := s3.NewFromConfig(awsCfg)
+	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketNameForSNS, s3Client,
 		"testdata/events-array.json",
 		"testdata/invalid.json",
 		"testdata/log.json",