diff --git a/x-pack/filebeat/docker-compose.yml b/x-pack/filebeat/docker-compose.yml
index d4733b8dd700..236a712ef07e 100644
--- a/x-pack/filebeat/docker-compose.yml
+++ b/x-pack/filebeat/docker-compose.yml
@@ -7,6 +7,7 @@ services:
     depends_on:
       elasticsearch: { condition: service_healthy }
       cometd: { condition: service_healthy }
+      localstack: { condition: service_healthy }
 
   elasticsearch:
     extends:
@@ -26,13 +27,12 @@ services:
       - 8080:8080
   
   localstack:
-    container_name: "${LOCALSTACK_DOCKER_NAME-localstack_main}"
     image: localstack/localstack
     hostname: localhost
     ports:
       - "4566:4566"            # LocalStack Gateway
     environment:
-      - SERVICES=sqs,sns,secretsmanager
+      - SERVICES=s3,sqs,sns,secretsmanager
       - DEBUG=1
       - DOCKER_HOST=unix:///var/run/docker.sock
       - HOST_TMP_FOLDER=${TMPDIR}
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/localstack.tf b/x-pack/filebeat/input/awss3/_meta/terraform/localstack.tf
new file mode 100644
index 000000000000..edef6fb67702
--- /dev/null
+++ b/x-pack/filebeat/input/awss3/_meta/terraform/localstack.tf
@@ -0,0 +1,90 @@
+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
+//  s3_force_path_style = 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/main.tf b/x-pack/filebeat/input/awss3/_meta/terraform/main.tf
index 8105a4beee06..fbfa0330e843 100644
--- a/x-pack/filebeat/input/awss3/_meta/terraform/main.tf
+++ b/x-pack/filebeat/input/awss3/_meta/terraform/main.tf
@@ -7,162 +7,139 @@ terraform {
   }
 }
 
-provider "aws" {
-  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" {
-  length  = 6
-  special = false
-  upper   = false
-}
-
-resource "aws_s3_bucket" "filebeat-integtest" {
-  bucket        = "filebeat-s3-integtest-${random_string.random.result}"
-  force_destroy = true
-}
-
-resource "aws_sqs_queue" "filebeat-integtest" {
-  name   = "filebeat-s3-integtest-${random_string.random.result}"
-  policy = <<POLICY
-{
-  "Version": "2012-10-17",
-  "Id": "sqspolicy",
-  "Statement": [
-    {
-      "Effect": "Allow",
-      "Principal": "*",
-      "Action": "sqs:SendMessage",
-      "Resource": "arn:aws:sqs:*:*:filebeat-s3-integtest-${random_string.random.result}",
-      "Condition": {
-        "ArnEquals": { "aws:SourceArn": "${aws_s3_bucket.filebeat-integtest.arn}" }
-      }
-    }
-  ]
-}
-POLICY
-
-  depends_on = [
-    aws_s3_bucket.filebeat-integtest,
-  ]
-}
-
-resource "aws_s3_bucket_notification" "bucket_notification" {
-  bucket = aws_s3_bucket.filebeat-integtest.id
-
-  queue {
-    queue_arn = aws_sqs_queue.filebeat-integtest.arn
-    events    = ["s3:ObjectCreated:*"]
-  }
-
-  depends_on = [
-    aws_s3_bucket.filebeat-integtest,
-    aws_sqs_queue.filebeat-integtest,
-  ]
-}
-
-resource "aws_sns_topic" "filebeat-integtest-sns" {
-  name = "filebeat-s3-integtest-sns-${random_string.random.result}"
-
-  policy = <<POLICY
-{
-    "Version":"2012-10-17",
-    "Statement":[{
-        "Effect": "Allow",
-        "Principal": { "Service": "s3.amazonaws.com" },
-        "Action": "SNS:Publish",
-        "Resource": "arn:aws:sns:*:*:filebeat-s3-integtest-sns-${random_string.random.result}",
-        "Condition":{
-            "ArnEquals": { "aws:SourceArn": "${aws_s3_bucket.filebeat-integtest-sns.arn}" }
-        }
-    }]
-}
-POLICY
-
-  depends_on = [
-    aws_s3_bucket.filebeat-integtest-sns,
-  ]
-}
-
-resource "aws_s3_bucket" "filebeat-integtest-sns" {
-  bucket        = "filebeat-s3-integtest-sns-${random_string.random.result}"
-  force_destroy = true
-}
-
-resource "aws_s3_bucket_notification" "bucket_notification-sns" {
-  bucket = aws_s3_bucket.filebeat-integtest-sns.id
-
-  topic {
-    topic_arn = aws_sns_topic.filebeat-integtest-sns.arn
-    events    = ["s3:ObjectCreated:*"]
-  }
-
-  depends_on = [
-    aws_s3_bucket.filebeat-integtest-sns,
-    aws_sns_topic.filebeat-integtest-sns,
-  ]
-}
-
-resource "aws_sqs_queue" "filebeat-integtest-sns" {
-  name = "filebeat-s3-integtest-sns-${random_string.random.result}"
-
-  policy = <<POLICY
-{
-  "Version": "2012-10-17",
-  "Statement": [
-    {
-      "Effect": "Allow",
-      "Principal": "*",
-      "Action": "sqs:SendMessage",
-      "Resource": "arn:aws:sqs:*:*:filebeat-s3-integtest-sns-${random_string.random.result}",
-      "Condition": {
-        "ArnEquals": { "aws:SourceArn": "${aws_sns_topic.filebeat-integtest-sns.arn}" }
-      }
-    }
-  ]
-}
-POLICY
-
-  depends_on = [
-    aws_s3_bucket.filebeat-integtest-sns,
-    aws_sns_topic.filebeat-integtest-sns
-  ]
-}
-
-resource "aws_sns_topic_subscription" "filebeat-integtest-sns" {
-  topic_arn = aws_sns_topic.filebeat-integtest-sns.arn
-  protocol  = "sqs"
-  endpoint  = aws_sqs_queue.filebeat-integtest-sns.arn
-}
+# provider "aws" {
+#   region = var.aws_region
+#   default_tags {
+#     tags = {
+#       environment  = var.ENVIRONMENT
+#       repo         = var.REPO
+#       branch       = var.BRANCH
+#       build        = var.BUILD_ID
+#       created_date = var.CREATED_DATE
+#     }
+#   }
+# }
+
+# resource "random_string" "random" {
+#   length  = 6
+#   special = false
+#   upper   = false
+# }
+
+# resource "aws_s3_bucket" "filebeat-integtest" {
+#   bucket        = "filebeat-s3-integtest-${random_string.random.result}"
+#   force_destroy = true
+# }
+
+# resource "aws_sqs_queue" "filebeat-integtest" {
+#   name   = "filebeat-s3-integtest-${random_string.random.result}"
+#   policy = <<POLICY
+# {
+#   "Version": "2012-10-17",
+#   "Id": "sqspolicy",
+#   "Statement": [
+#     {
+#       "Effect": "Allow",
+#       "Principal": "*",
+#       "Action": "sqs:SendMessage",
+#       "Resource": "arn:aws:sqs:*:*:filebeat-s3-integtest-${random_string.random.result}",
+#       "Condition": {
+#         "ArnEquals": { "aws:SourceArn": "${aws_s3_bucket.filebeat-integtest.arn}" }
+#       }
+#     }
+#   ]
+# }
+# POLICY
+
+#   depends_on = [
+#     aws_s3_bucket.filebeat-integtest,
+#   ]
+# }
+
+# resource "aws_s3_bucket_notification" "bucket_notification" {
+#   bucket = aws_s3_bucket.filebeat-integtest.id
+
+#   queue {
+#     queue_arn = aws_sqs_queue.filebeat-integtest.arn
+#     events    = ["s3:ObjectCreated:*"]
+#   }
+
+#   depends_on = [
+#     aws_s3_bucket.filebeat-integtest,
+#     aws_sqs_queue.filebeat-integtest,
+#   ]
+# }
+
+# resource "aws_sns_topic" "filebeat-integtest-sns" {
+#   name = "filebeat-s3-integtest-sns-${random_string.random.result}"
+
+#   policy = <<POLICY
+# {
+#     "Version":"2012-10-17",
+#     "Statement":[{
+#         "Effect": "Allow",
+#         "Principal": { "Service": "s3.amazonaws.com" },
+#         "Action": "SNS:Publish",
+#         "Resource": "arn:aws:sns:*:*:filebeat-s3-integtest-sns-${random_string.random.result}",
+#         "Condition":{
+#             "ArnEquals": { "aws:SourceArn": "${aws_s3_bucket.filebeat-integtest-sns.arn}" }
+#         }
+#     }]
+# }
+# POLICY
+
+#   depends_on = [
+#     aws_s3_bucket.filebeat-integtest-sns,
+#   ]
+# }
+
+# resource "aws_s3_bucket" "filebeat-integtest-sns" {
+#   bucket        = "filebeat-s3-integtest-sns-${random_string.random.result}"
+#   force_destroy = true
+# }
+
+# resource "aws_s3_bucket_notification" "bucket_notification-sns" {
+#   bucket = aws_s3_bucket.filebeat-integtest-sns.id
+
+#   topic {
+#     topic_arn = aws_sns_topic.filebeat-integtest-sns.arn
+#     events    = ["s3:ObjectCreated:*"]
+#   }
+
+#   depends_on = [
+#     aws_s3_bucket.filebeat-integtest-sns,
+#     aws_sns_topic.filebeat-integtest-sns,
+#   ]
+# }
+
+# resource "aws_sqs_queue" "filebeat-integtest-sns" {
+#   name = "filebeat-s3-integtest-sns-${random_string.random.result}"
+
+#   policy = <<POLICY
+# {
+#   "Version": "2012-10-17",
+#   "Statement": [
+#     {
+#       "Effect": "Allow",
+#       "Principal": "*",
+#       "Action": "sqs:SendMessage",
+#       "Resource": "arn:aws:sqs:*:*:filebeat-s3-integtest-sns-${random_string.random.result}",
+#       "Condition": {
+#         "ArnEquals": { "aws:SourceArn": "${aws_sns_topic.filebeat-integtest-sns.arn}" }
+#       }
+#     }
+#   ]
+# }
+# POLICY
+
+#   depends_on = [
+#     aws_s3_bucket.filebeat-integtest-sns,
+#     aws_sns_topic.filebeat-integtest-sns
+#   ]
+# }
+
+# resource "aws_sns_topic_subscription" "filebeat-integtest-sns" {
+#   topic_arn = aws_sns_topic.filebeat-integtest-sns.arn
+#   protocol  = "sqs"
+#   endpoint  = aws_sqs_queue.filebeat-integtest-sns.arn
+# }
diff --git a/x-pack/filebeat/input/awss3/_meta/terraform/outputs.tf b/x-pack/filebeat/input/awss3/_meta/terraform/outputs.tf
index a9851024bb05..f66d9c7a468e 100644
--- a/x-pack/filebeat/input/awss3/_meta/terraform/outputs.tf
+++ b/x-pack/filebeat/input/awss3/_meta/terraform/outputs.tf
@@ -1,11 +1,21 @@
-resource "local_file" "secrets" {
+# resource "local_file" "secrets" {
+#   content = yamlencode({
+#     "queue_url" : aws_sqs_queue.filebeat-integtest.url
+#     "aws_region" : aws_s3_bucket.filebeat-integtest.region
+#     "bucket_name" : aws_s3_bucket.filebeat-integtest.id
+#     "bucket_name_for_sns" : aws_s3_bucket.filebeat-integtest-sns.id
+#     "queue_url_for_sns" : aws_sqs_queue.filebeat-integtest-sns.url
+#   })
+#   filename        = "${path.module}/outputs.yml"
+#   file_permission = "0644"
+# }
+
+resource "local_file" "secrets-localstack" {
   content = yamlencode({
-    "queue_url" : aws_sqs_queue.filebeat-integtest.url
-    "aws_region" : aws_s3_bucket.filebeat-integtest.region
-    "bucket_name" : aws_s3_bucket.filebeat-integtest.id
-    "bucket_name_for_sns" : aws_s3_bucket.filebeat-integtest-sns.id
-    "queue_url_for_sns" : aws_sqs_queue.filebeat-integtest-sns.url
+    "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.yml"
+  filename        = "${path.module}/outputs-localstack.yml"
   file_permission = "0644"
 }
diff --git a/x-pack/filebeat/input/awss3/config.go b/x-pack/filebeat/input/awss3/config.go
index ab6daad2b0c7..65fde0d34f7c 100644
--- a/x-pack/filebeat/input/awss3/config.go
+++ b/x-pack/filebeat/input/awss3/config.go
@@ -100,7 +100,7 @@ func (c *config) Validate() error {
 	}
 
 	if c.AWSConfig.FIPSEnabled && c.NonAWSBucketName != "" {
-		return errors.New("fips_enabled cannot be used with a non-AWS S3 bucket.")
+		return errors.New("fips_enabled cannot be used with a non-AWS S3 bucket")
 	}
 	if c.PathStyle && c.NonAWSBucketName == "" {
 		return errors.New("path_style can only be used when polling non-AWS S3 services")
diff --git a/x-pack/filebeat/input/awss3/input.go b/x-pack/filebeat/input/awss3/input.go
index 2d2eea3ded52..0225b10b641c 100644
--- a/x-pack/filebeat/input/awss3/input.go
+++ b/x-pack/filebeat/input/awss3/input.go
@@ -189,8 +189,9 @@ func (in *s3Input) createSQSReceiver(ctx v2.Context, pipeline beat.Pipeline) (*s
 		client: s3.NewFromConfig(in.awsConfig, func(o *s3.Options) {
 			if in.config.AWSConfig.FIPSEnabled {
 				o.EndpointOptions.UseFIPSEndpoint = awssdk.FIPSEndpointStateEnabled
-				o.UsePathStyle = in.config.PathStyle
 			}
+			o.UsePathStyle = in.config.PathStyle
+
 		}),
 	}
 
@@ -248,34 +249,28 @@ func (in *s3Input) createS3Lister(ctx v2.Context, cancelCtx context.Context, cli
 		if in.config.AWSConfig.FIPSEnabled {
 			o.EndpointOptions.UseFIPSEndpoint = awssdk.FIPSEndpointStateEnabled
 		}
-		o.UsePathStyle = true
+		o.UsePathStyle = in.config.PathStyle
 	})
+	regionName, err := getRegionForBucket(cancelCtx, s3Client, bucketName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get AWS region for bucket: %w", err)
+	}
 
-	// regionName, err := getRegionForBucket(cancelCtx, s3Client, bucketName)
-	// if err != nil {
-	// 	return nil, fmt.Errorf("failed to get AWS region for bucket: %w", err)
-	// }
-
-	// originalAwsConfigRegion := in.awsConfig.Region
-
-	// ctx.Logger.Errorf("**ORIGINAL*** %s", originalAwsConfigRegion)
-
-	// in.awsConfig.Region = regionName
-
-	// ctx.Logger.Errorf("***** %s", regionName)
+	originalAwsConfigRegion := in.awsConfig.Region
+	in.awsConfig.Region = regionName
 
-	// if regionName != originalAwsConfigRegion {
-	// 	s3Client = s3.NewFromConfig(in.awsConfig, func(o *s3.Options) {
-	// 		if in.config.NonAWSBucketName != "" {
-	// 			o.EndpointResolver = nonAWSBucketResolver{endpoint: in.config.AWSConfig.Endpoint}
-	// 		}
+	if regionName != originalAwsConfigRegion {
+		s3Client = s3.NewFromConfig(in.awsConfig, func(o *s3.Options) {
+			if in.config.NonAWSBucketName != "" {
+				o.EndpointResolver = nonAWSBucketResolver{endpoint: in.config.AWSConfig.Endpoint}
+			}
 
-	// 		if in.config.AWSConfig.FIPSEnabled {
-	// 			o.EndpointOptions.UseFIPSEndpoint = awssdk.FIPSEndpointStateEnabled
-	// 		}
-	// 		o.UsePathStyle = true
-	// 	})
-	// }
+			if in.config.AWSConfig.FIPSEnabled {
+				o.EndpointOptions.UseFIPSEndpoint = awssdk.FIPSEndpointStateEnabled
+			}
+			o.UsePathStyle = in.config.PathStyle
+		})
+	}
 
 	s3API := &awsS3API{
 		client: s3Client,
diff --git a/x-pack/filebeat/input/awss3/input_integration_test.go b/x-pack/filebeat/input/awss3/input_integration_test.go
index 773b74493c04..3afa5456f343 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,11 +52,18 @@ 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))
 	}
@@ -65,7 +73,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)
 	}
@@ -144,16 +151,132 @@ 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,
+		BucketARN:           "",
+		NonAWSBucketName:    "",
+	}
+	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 into S3
+// and a notification is generated to SQS using 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)
+	}
+
+	// Initialize s3Input with the test config
+	s3Input := &s3Input{
+		config:    config,
+		awsConfig: awsCfg,
+		store:     openTestStatestore(),
+	}
+
+	// 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()
+	})
+
+	// 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.sqsMessagesWaiting.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.Greater(t, s3Input.metrics.sqsWorkerUtilization.Get(), 0.0)
+}
+
 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 +321,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 +361,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 +395,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 +443,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 +461,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 +482,6 @@ func TestPaginatorListPrefix(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	s3Client := s3.NewFromConfig(cfg)
-
 	s3API := &awsS3API{
 		client: s3Client,
 	}
@@ -387,12 +509,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.BucketName, s3Client,
 		"testdata/events-array.json",
 		"testdata/invalid.json",
 		"testdata/log.json",
diff --git a/x-pack/filebeat/input/awss3/input_localstack_integration_test.go b/x-pack/filebeat/input/awss3/input_localstack_integration_test.go
deleted file mode 100644
index 34507b616ced..000000000000
--- a/x-pack/filebeat/input/awss3/input_localstack_integration_test.go
+++ /dev/null
@@ -1,490 +0,0 @@
-// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
-// or more contributor license agreements. Licensed under the Elastic License;
-// you may not use this file except in compliance with the Elastic License.
-
-// See _meta/terraform/README.md for integration test usage instructions.
-
-//go:build integration
-
-package awss3
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"io/ioutil"
-	"os"
-	"path"
-	"path/filepath"
-	"runtime"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/aws/aws-sdk-go-v2/aws"
-	awsConfig "github.com/aws/aws-sdk-go-v2/config"
-	s3manager "github.com/aws/aws-sdk-go-v2/feature/s3/manager"
-	"github.com/aws/aws-sdk-go-v2/service/s3"
-	"github.com/aws/aws-sdk-go-v2/service/sqs"
-	"github.com/stretchr/testify/assert"
-	"golang.org/x/sync/errgroup"
-	"gopkg.in/yaml.v2"
-
-	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
-	conf "github.com/elastic/elastic-agent-libs/config"
-	"github.com/elastic/elastic-agent-libs/logp"
-)
-
-const (
-	inputID = "test_id"
-)
-
-const (
-	terraformOutputYML = "_meta/terraform/outputs.yml"
-)
-
-type terraformOutputData struct {
-	AWSRegion        string `yaml:"aws_region"`
-	BucketName       string `yaml:"bucket_name"`
-	QueueURL         string `yaml:"queue_url"`
-	BucketNameForSNS string `yaml:"bucket_name_for_sns"`
-	QueueURLForSNS   string `yaml:"queue_url_for_sns"`
-}
-
-func getTerraformOutputs(t *testing.T) terraformOutputData {
-	t.Helper()
-
-	_, filename, _, _ := runtime.Caller(0)
-	ymlData, err := ioutil.ReadFile(path.Join(path.Dir(filename), terraformOutputYML))
-	if os.IsNotExist(err) {
-		t.Skipf("Run 'terraform apply' in %v to setup S3 and SQS for the test.", filepath.Dir(terraformOutputYML))
-	}
-	if err != nil {
-		t.Fatalf("failed reading terraform output data: %v", err)
-	}
-
-	var rtn terraformOutputData
-	dec := yaml.NewDecoder(bytes.NewReader(ymlData))
-	dec.SetStrict(true)
-	if err = dec.Decode(&rtn); err != nil {
-		t.Fatal(err)
-	}
-
-	return rtn
-}
-
-func makeTestConfigS3(s3bucket string) *conf.C {
-	return conf.MustNewConfigFrom(fmt.Sprintf(`---
-bucket_arn: aws:s3:::%s
-region: us-east-1
-number_of_workers: 2
-file_selectors:
--
-  regex: 'events-array.json$'
-  expand_event_list_from_field: Events
-  include_s3_metadata:
-    - last-modified
-    - x-amz-version-id
-    - x-amz-storage-class
-    - Content-Length
-    - Content-Type
--
-  regex: '\.(?:nd)?json(\.gz)?$'
--
-  regex: 'multiline.txt$'
-  parsers:
-    - multiline:
-        pattern: "^<Event"
-        negate:  true
-        match:   after
-`, s3bucket))
-}
-
-func makeTestConfigSQS(queueURL string) *conf.C {
-	return conf.MustNewConfigFrom(fmt.Sprintf(`---
-queue_url: %s
-region: us-east-1
-max_number_of_messages: 1
-visibility_timeout: 30s
-file_selectors:
--
-  regex: 'events-array.json$'
-  expand_event_list_from_field: Events
-  include_s3_metadata:
-    - last-modified
-    - x-amz-version-id
-    - x-amz-storage-class
-    - Content-Length
-    - Content-Type
--
-  regex: '\.(?:nd)?json(\.gz)?$'
--
-  regex: 'multiline.txt$'
-  parsers:
-    - multiline:
-        pattern: "^<Event"
-        negate:  true
-        match:   after
-`, queueURL))
-}
-
-func createInput(t *testing.T, cfg *conf.C) *s3Input {
-	inputV2, err := Plugin(openTestStatestore()).Manager.Create(cfg)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	return inputV2.(*s3Input)
-}
-
-func newV2Context() (v2.Context, func()) {
-	ctx, cancel := context.WithCancel(context.Background())
-	return v2.Context{
-		Logger:      logp.NewLogger(inputName).With("id", inputID),
-		ID:          inputID,
-		Cancelation: ctx,
-	}, cancel
-}
-
-func TestInputRunSQS(t *testing.T) {
-	logp.TestingSetup()
-
-	// Terraform is used to set up S3 and SQS and must be executed manually.
-	tfConfig := getTerraformOutputs(t)
-
-	// Ensure SQS is empty before testing.
-	drainSQS(t, tfConfig.AWSRegion, tfConfig.QueueURL)
-
-	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName,
-		"testdata/test.xml", // Skipped (no match).
-	)
-
-	s3Input := createInput(t, makeTestConfigSQS(tfConfig.QueueURL))
-
-	inputCtx, cancel := newV2Context()
-	t.Cleanup(cancel)
-	time.AfterFunc(15*time.Second, func() {
-		cancel()
-	})
-
-	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(), 1) // 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.sqsMessagesWaiting.Get(), 0)
-	assert.EqualValues(t, s3Input.metrics.s3ObjectsInflight.Get(), 0)
-	//assert.EqualValues(t, s3Input.metrics.s3ObjectsRequestedTotal.Get(), 7)
-	//assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), 12)
-	//assert.Greater(t, s3Input.metrics.sqsLagTime.Mean(), 0.0)
-	//assert.Greater(t, s3Input.metrics.sqsWorkerUtilization.Get(), 0.0)
-}
-
-func TestInputRunS3(t *testing.T) {
-	logp.TestingSetup()
-
-	// Terraform is used to set up S3 and must be executed manually.
-	tfConfig := getTerraformOutputs(t)
-
-	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName,
-		"testdata/test.xml", // Skipped (no match).
-	)
-	fmt.Println(tfConfig.BucketName)
-
-	s3Input := createInput(t, makeTestConfigS3(tfConfig.BucketName))
-
-	inputCtx, cancel := newV2Context()
-	t.Cleanup(cancel)
-	time.AfterFunc(15*time.Second, func() {
-		cancel()
-	})
-
-	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.s3ObjectsInflight.Get(), 0)
-	assert.EqualValues(t, s3Input.metrics.s3ObjectsRequestedTotal.Get(), 0)
-	assert.EqualValues(t, s3Input.metrics.s3ObjectsListedTotal.Get(), 0)
-	assert.EqualValues(t, s3Input.metrics.s3ObjectsProcessedTotal.Get(), 0)
-	assert.EqualValues(t, s3Input.metrics.s3ObjectsAckedTotal.Get(), 0)
-	assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), 0)
-}
-
-func uploadS3TestFiles(t *testing.T, region, bucket string, filenames ...string) {
-	t.Helper()
-	awsEndpoint := "http://localhost:4566"
-	awsRegion := "us-east-1"
-
-	customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
-		if awsEndpoint != "" {
-			return aws.Endpoint{
-				PartitionID:   "aws",
-				URL:           awsEndpoint,
-				SigningRegion: awsRegion,
-			}, nil
-		}
-
-		// returning EndpointNotFoundError will allow the service to fallback to its default resolution
-		return aws.Endpoint{}, &aws.EndpointNotFoundError{}
-	})
-
-	cfg, err := awsConfig.LoadDefaultConfig(context.TODO(),
-		awsConfig.WithRegion(awsRegion),
-		awsConfig.WithEndpointResolverWithOptions(customResolver),
-	)
-
-	if err != nil {
-		t.Fatal(err)
-	}
-	cfg.Region = region
-	s3Client := s3.NewFromConfig(cfg, func(o *s3.Options) {
-		o.UsePathStyle = true
-	})
-	uploader := s3manager.NewUploader(s3Client)
-
-	_, basefile, _, _ := runtime.Caller(0)
-	basedir := path.Dir(basefile)
-	for _, filename := range filenames {
-		data, err := ioutil.ReadFile(path.Join(basedir, filename))
-		if err != nil {
-			t.Fatalf("Failed to open file %q, %v", filename, err)
-		}
-
-		contentType := ""
-		if strings.HasSuffix(filename, "ndjson") || strings.HasSuffix(filename, "ndjson.gz") {
-			contentType = contentTypeNDJSON + "; charset=UTF-8"
-		} else if strings.HasSuffix(filename, "json") || strings.HasSuffix(filename, "json.gz") {
-			contentType = contentTypeJSON + "; charset=UTF-8"
-		}
-
-		// Upload the file to S3.
-		result, err := uploader.Upload(context.Background(), &s3.PutObjectInput{
-			Bucket:      aws.String(bucket),
-			Key:         aws.String(filepath.Base(filename)),
-			Body:        bytes.NewReader(data),
-			ContentType: aws.String(contentType),
-		})
-		if err != nil {
-			t.Fatalf("Failed to upload file %q: %v", filename, err)
-		}
-		t.Logf("File uploaded to %s", result.Location)
-	}
-}
-
-func drainSQS(t *testing.T, region string, queueURL string) {
-	awsEndpoint := "http://localhost:4566"
-	awsRegion := "us-east-1"
-
-	customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
-		if awsEndpoint != "" {
-			return aws.Endpoint{
-				PartitionID:   "aws",
-				URL:           awsEndpoint,
-				SigningRegion: awsRegion,
-			}, nil
-		}
-
-		// returning EndpointNotFoundError will allow the service to fallback to its default resolution
-		return aws.Endpoint{}, &aws.EndpointNotFoundError{}
-	})
-
-	cfg, err := awsConfig.LoadDefaultConfig(context.TODO(),
-		awsConfig.WithRegion(awsRegion),
-		awsConfig.WithEndpointResolverWithOptions(customResolver),
-	)
-
-	if err != nil {
-		t.Fatal(err)
-	}
-	cfg.Region = region
-
-	sqs := &awsSQSAPI{
-		client:            sqs.NewFromConfig(cfg),
-		queueURL:          queueURL,
-		apiTimeout:        1 * time.Minute,
-		visibilityTimeout: 30 * time.Second,
-		longPollWaitTime:  10,
-	}
-
-	ctx := context.Background()
-	var deletedCount int
-	for {
-		msgs, err := sqs.ReceiveMessage(ctx, 10)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if len(msgs) == 0 {
-			break
-		}
-
-		for _, msg := range msgs {
-			if err = sqs.DeleteMessage(ctx, &msg); err != nil {
-				t.Fatal(err)
-			}
-			deletedCount++
-		}
-	}
-	t.Logf("Drained %d SQS messages.", deletedCount)
-}
-
-func TestGetBucketNameFromARN(t *testing.T) {
-	bucketName := getBucketNameFromARN("arn:aws:s3:::my_corporate_bucket")
-	assert.Equal(t, "my_corporate_bucket", bucketName)
-}
-
-func TestGetRegionForBucketARN(t *testing.T) {
-	logp.TestingSetup()
-
-	// Terraform is used to set up S3 and must be executed manually.
-	tfConfig := getTerraformOutputs(t)
-
-	awsEndpoint := "http://localhost:4566"
-	awsRegion := "us-east-1"
-
-	customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
-		if awsEndpoint != "" {
-			return aws.Endpoint{
-				PartitionID:   "aws",
-				URL:           awsEndpoint,
-				SigningRegion: awsRegion,
-			}, nil
-		}
-
-		// returning EndpointNotFoundError will allow the service to fallback to its default resolution
-		return aws.Endpoint{}, &aws.EndpointNotFoundError{}
-	})
-
-	cfg, err := awsConfig.LoadDefaultConfig(context.TODO(),
-		awsConfig.WithRegion(awsRegion),
-		awsConfig.WithEndpointResolverWithOptions(customResolver),
-	)
-
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	s3Client := s3.NewFromConfig(cfg)
-
-	regionName, err := getRegionForBucket(context.Background(), s3Client, getBucketNameFromARN(tfConfig.BucketName))
-	assert.NoError(t, err)
-	assert.Equal(t, tfConfig.AWSRegion, regionName)
-}
-
-// func TestPaginatorListPrefix(t *testing.T) {
-// 	logp.TestingSetup()
-
-// 	// Terraform is used to set up S3 and must be executed manually.
-// 	tfConfig := getTerraformOutputs(t)
-
-// 	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketName,
-// 		"testdata/test.xml", // Skipped (no match).
-// 	)
-
-// 	awsEndpoint := "http://localhost:4566"
-// 	awsRegion := "us-east-1"
-
-// 	customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
-// 		if awsEndpoint != "" {
-// 			return aws.Endpoint{
-// 				PartitionID:   "aws",
-// 				URL:           awsEndpoint,
-// 				SigningRegion: awsRegion,
-// 			}, nil
-// 		}
-
-// 		// returning EndpointNotFoundError will allow the service to fallback to its default resolution
-// 		return aws.Endpoint{}, &aws.EndpointNotFoundError{}
-// 	})
-
-// 	cfg, err := awsConfig.LoadDefaultConfig(context.TODO(),
-// 		awsConfig.WithRegion(awsRegion),
-// 		awsConfig.WithEndpointResolverWithOptions(customResolver),
-// 	)
-
-// 	cfg.Region = tfConfig.AWSRegion
-// 	if err != nil {
-// 		t.Fatal(err)
-// 	}
-
-// 	s3Client := s3.NewFromConfig(cfg)
-
-// 	s3API := &awsS3API{
-// 		client: s3Client,
-// 	}
-
-// 	var objects []string
-// 	paginator := s3API.ListObjectsPaginator(tfConfig.BucketName, "log")
-// 	for paginator.HasMorePages() {
-// 		page, err := paginator.NextPage(context.Background())
-// 		assert.NoError(t, err)
-// 		for _, object := range page.Contents {
-// 			objects = append(objects, *object.Key)
-// 		}
-// 	}
-
-// 	expected := []string{
-// 		"log.json",
-// 		"log.ndjson",
-// 		"log.txt",
-// 	}
-
-// 	assert.Equal(t, expected, objects)
-// }
-
-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)
-
-	// Ensure SQS is empty before testing.
-	drainSQS(t, tfConfig.AWSRegion, tfConfig.QueueURLForSNS)
-
-	uploadS3TestFiles(t, tfConfig.AWSRegion, tfConfig.BucketNameForSNS,
-		"testdata/test.xml", // Skipped (no match).
-	)
-
-	s3Input := createInput(t, makeTestConfigSQS(tfConfig.QueueURLForSNS))
-
-	inputCtx, cancel := newV2Context()
-	t.Cleanup(cancel)
-	time.AfterFunc(15*time.Second, func() {
-		cancel()
-	})
-
-	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(), 0) // S3 could batch notifications.
-	assert.EqualValues(t, s3Input.metrics.sqsMessagesInflight.Get(), 0)
-	assert.EqualValues(t, s3Input.metrics.sqsMessagesDeletedTotal.Get(), 0)
-	assert.EqualValues(t, s3Input.metrics.sqsMessagesReturnedTotal.Get(), 0) // Invalid JSON is returned so that it can eventually be DLQed.
-	// assert.EqualValues(t, s3Input.metrics.sqsVisibilityTimeoutExtensionsTotal.Get(), 0)
-	// assert.EqualValues(t, s3Input.metrics.sqsMessagesWaiting.Get(), 0)
-	// assert.EqualValues(t, s3Input.metrics.s3ObjectsInflight.Get(), 0)
-	// assert.EqualValues(t, s3Input.metrics.s3ObjectsRequestedTotal.Get(), 7)
-	// assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), 12)
-	// assert.Greater(t, s3Input.metrics.sqsLagTime.Mean(), 0.0)
-	// assert.Greater(t, s3Input.metrics.sqsWorkerUtilization.Get(), 0.0)
-}
diff --git a/x-pack/filebeat/input/awss3/testdata/test.xml b/x-pack/filebeat/input/awss3/testdata/test.xml
index c92b65a8660d..ce023bbf68d8 100644
--- a/x-pack/filebeat/input/awss3/testdata/test.xml
+++ b/x-pack/filebeat/input/awss3/testdata/test.xml
@@ -1,2 +1,6 @@
-<head>
-</head>
\ No newline at end of file
+<note>
+    <to>bar</to>
+    <from>foo</from>
+    <heading>Reminder</heading>
+    <body>Don't forget me this weekend!</body>
+</note>
\ No newline at end of file