diff --git a/.changelog/41082.txt b/.changelog/41082.txt
new file mode 100644
index 00000000000..7fc1fcf8e14
--- /dev/null
+++ b/.changelog/41082.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_pipes_pipe: Add `kms_key_identifier` argument
+```
\ No newline at end of file
diff --git a/internal/service/pipes/pipe.go b/internal/service/pipes/pipe.go
index 57c2b564283..6b293ad3786 100644
--- a/internal/service/pipes/pipe.go
+++ b/internal/service/pipes/pipe.go
@@ -74,7 +74,11 @@ func resourcePipe() *schema.Resource {
 					ValidateFunc: verify.ValidARN,
 				},
 				"enrichment_parameters": enrichmentParametersSchema(),
-				"log_configuration":     logConfigurationSchema(),
+				"kms_key_identifier": {
+					Type:     schema.TypeString,
+					Optional: true,
+				},
+				"log_configuration": logConfigurationSchema(),
 				names.AttrName: {
 					Type:          schema.TypeString,
 					Optional:      true,
@@ -155,6 +159,10 @@ func resourcePipeCreate(ctx context.Context, d *schema.ResourceData, meta interf
 		input.EnrichmentParameters = expandPipeEnrichmentParameters(v.([]interface{})[0].(map[string]interface{}))
 	}
 
+	if v, ok := d.GetOk("kms_key_identifier"); ok && v != "" {
+		input.KmsKeyIdentifier = aws.String(v.(string))
+	}
+
 	if v, ok := d.GetOk("source_parameters"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
 		input.SourceParameters = expandPipeSourceParameters(v.([]interface{})[0].(map[string]interface{}))
 	}
@@ -209,6 +217,7 @@ func resourcePipeRead(ctx context.Context, d *schema.ResourceData, meta interfac
 	} else {
 		d.Set("enrichment_parameters", nil)
 	}
+	d.Set("kms_key_identifier", output.KmsKeyIdentifier)
 	if v := output.LogConfiguration; !types.IsZero(v) {
 		if err := d.Set("log_configuration", []interface{}{flattenPipeLogConfiguration(v)}); err != nil {
 			return sdkdiag.AppendErrorf(diags, "setting log_configuration: %s", err)
@@ -262,6 +271,10 @@ func resourcePipeUpdate(ctx context.Context, d *schema.ResourceData, meta interf
 			}
 		}
 
+		if d.HasChange("kms_key_identifier") {
+			input.KmsKeyIdentifier = aws.String(d.Get("kms_key_identifier").(string))
+		}
+
 		if d.HasChange("log_configuration") {
 			if v, ok := d.GetOk("log_configuration"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
 				input.LogConfiguration = expandPipeLogConfigurationParameters(v.([]interface{})[0].(map[string]interface{}))
diff --git a/internal/service/pipes/pipe_test.go b/internal/service/pipes/pipe_test.go
index ed7240b270e..36bd0adca72 100644
--- a/internal/service/pipes/pipe_test.go
+++ b/internal/service/pipes/pipe_test.go
@@ -49,6 +49,7 @@ func TestAccPipesPipe_basicSQS(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "desired_state", "RUNNING"),
 					resource.TestCheckResourceAttr(resourceName, "enrichment", ""),
 					resource.TestCheckResourceAttr(resourceName, "enrichment_parameters.#", "0"),
+					resource.TestCheckResourceAttr(resourceName, "kms_key_identifier", ""),
 					resource.TestCheckResourceAttr(resourceName, names.AttrName, rName),
 					resource.TestCheckResourceAttrPair(resourceName, names.AttrRoleARN, "aws_iam_role.test", names.AttrARN),
 					resource.TestCheckResourceAttrPair(resourceName, names.AttrSource, "aws_sqs_queue.source", names.AttrARN),
@@ -312,6 +313,55 @@ func TestAccPipesPipe_enrichmentParameters(t *testing.T) {
 	})
 }
 
+func TestAccPipesPipe_kmsKeyIdentifier(t *testing.T) {
+	ctx := acctest.Context(t)
+	var pipe pipes.DescribePipeOutput
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_pipes_pipe.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck: func() {
+			acctest.PreCheck(ctx, t)
+			acctest.PreCheckPartitionHasService(t, names.PipesEndpointID)
+			testAccPreCheck(ctx, t)
+		},
+		ErrorCheck:               acctest.ErrorCheck(t, names.PipesServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckPipeDestroy(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccPipeConfig_kmsKeyIdentifier(rName, "${aws_kms_key.test_1.id}"),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckPipeExists(ctx, resourceName, &pipe),
+					acctest.MatchResourceAttrRegionalARN(ctx, resourceName, names.AttrARN, "pipes", regexache.MustCompile(regexp.QuoteMeta(`pipe/`+rName))),
+					resource.TestCheckResourceAttrPair(resourceName, "kms_key_identifier", "aws_kms_key.test_1", names.AttrID),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccPipeConfig_kmsKeyIdentifier(rName, "${aws_kms_key.test_2.arn}"),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckPipeExists(ctx, resourceName, &pipe),
+					acctest.MatchResourceAttrRegionalARN(ctx, resourceName, names.AttrARN, "pipes", regexache.MustCompile(regexp.QuoteMeta(`pipe/`+rName))),
+					resource.TestCheckResourceAttrPair(resourceName, "kms_key_identifier", "aws_kms_key.test_2", names.AttrARN),
+				),
+			},
+			{
+				Config: testAccPipeConfig_kmsKeyIdentifier(rName, ""),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckPipeExists(ctx, resourceName, &pipe),
+					acctest.MatchResourceAttrRegionalARN(ctx, resourceName, names.AttrARN, "pipes", regexache.MustCompile(regexp.QuoteMeta(`pipe/`+rName))),
+					resource.TestCheckResourceAttr(resourceName, "kms_key_identifier", ""),
+				),
+			},
+		},
+	})
+}
+
 func TestAccPipesPipe_logConfiguration_cloudwatchLogsLogDestination(t *testing.T) {
 	ctx := acctest.Context(t)
 	var pipe pipes.DescribePipeOutput
@@ -2058,6 +2108,27 @@ resource "aws_pipes_pipe" "test" {
 `, rName))
 }
 
+func testAccPipeConfig_kmsKeyIdentifier(rName, kmsKeyID string) string {
+	return acctest.ConfigCompose(
+		testAccPipeConfig_base(rName),
+		testAccPipeConfig_baseSQSSource(rName),
+		testAccPipeConfig_baseSQSTarget(rName),
+		fmt.Sprintf(`
+resource "aws_kms_key" "test_1" {}
+resource "aws_kms_key" "test_2" {}
+
+resource "aws_pipes_pipe" "test" {
+  depends_on = [aws_iam_role_policy.source, aws_iam_role_policy.target]
+
+  kms_key_identifier = %[2]q
+  name               = %[1]q
+  role_arn           = aws_iam_role.test.arn
+  source             = aws_sqs_queue.source.arn
+  target             = aws_sqs_queue.target.arn
+}
+`, rName, kmsKeyID))
+}
+
 func testAccPipeConfig_logConfiguration_cloudwatchLogsLogDestination(rName string) string {
 	return acctest.ConfigCompose(
 		testAccPipeConfig_base(rName),
diff --git a/website/docs/r/pipes_pipe.html.markdown b/website/docs/r/pipes_pipe.html.markdown
index 5598a886003..8d95062f2d0 100644
--- a/website/docs/r/pipes_pipe.html.markdown
+++ b/website/docs/r/pipes_pipe.html.markdown
@@ -205,6 +205,7 @@ The following arguments are optional:
 * `desired_state` - (Optional) The state the pipe should be in. One of: `RUNNING`, `STOPPED`.
 * `enrichment` - (Optional) Enrichment resource of the pipe (typically an ARN). Read more about enrichment in the [User Guide](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html#pipes-enrichment).
 * `enrichment_parameters` - (Optional) Parameters to configure enrichment for your pipe. Detailed below.
+* `kms_key_identifier` - (Optional) Identifier of the AWS KMS customer managed key for EventBridge to use, if you choose to use a customer managed key to encrypt pipe data. The identifier can be the key Amazon Resource Name (ARN), KeyId, key alias, or key alias ARN. If not set, EventBridge uses an AWS owned key to encrypt pipe data.
 * `log_configuration` - (Optional) Logging configuration settings for the pipe. Detailed below.
 * `name` - (Optional) Name of the pipe. If omitted, Terraform will assign a random, unique name. Conflicts with `name_prefix`.
 * `name_prefix` - (Optional) Creates a unique name beginning with the specified prefix. Conflicts with `name`.