diff --git a/.changelog/39212.txt b/.changelog/39212.txt
new file mode 100644
index 00000000000..366841ddb26
--- /dev/null
+++ b/.changelog/39212.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+data-source/aws_kinesis_stream: Add `encryption_type` and `kms_key_id` attributes
+```
\ No newline at end of file
diff --git a/internal/service/kinesis/stream_data_source.go b/internal/service/kinesis/stream_data_source.go
index c5cc284d1c9..6b549cdea6c 100644
--- a/internal/service/kinesis/stream_data_source.go
+++ b/internal/service/kinesis/stream_data_source.go
@@ -36,6 +36,14 @@ func DataSourceStream() *schema.Resource {
 				Type:     schema.TypeInt,
 				Computed: true,
 			},
+			"encryption_type": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			names.AttrKMSKeyID: {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
 			names.AttrName: {
 				Type:     schema.TypeString,
 				Required: true,
@@ -115,6 +123,8 @@ func dataSourceStreamRead(ctx context.Context, d *schema.ResourceData, meta inte
 	d.Set(names.AttrARN, stream.StreamARN)
 	d.Set("closed_shards", aws.ToStringSlice(closedShards))
 	d.Set("creation_timestamp", aws.ToTime(stream.StreamCreationTimestamp).Unix())
+	d.Set("encryption_type", stream.EncryptionType)
+	d.Set(names.AttrKMSKeyID, stream.KeyId)
 	d.Set(names.AttrName, stream.StreamName)
 	d.Set("open_shards", aws.ToStringSlice(openShards))
 	d.Set(names.AttrRetentionPeriod, stream.RetentionPeriodHours)
diff --git a/internal/service/kinesis/stream_data_source_test.go b/internal/service/kinesis/stream_data_source_test.go
index 3b78d8aab67..dc2793c1bfc 100644
--- a/internal/service/kinesis/stream_data_source_test.go
+++ b/internal/service/kinesis/stream_data_source_test.go
@@ -50,6 +50,35 @@ func TestAccKinesisStreamDataSource_basic(t *testing.T) {
 	})
 }
 
+func TestAccKinesisStreamDataSource_encryption(t *testing.T) {
+	ctx := acctest.Context(t)
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	dataSourceName := "data.aws_kinesis_stream.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.KinesisServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckStreamDestroy(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccStreamDataSourceConfig_encryption(rName, 2),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttrSet(dataSourceName, names.AttrARN),
+					resource.TestCheckResourceAttrSet(dataSourceName, "creation_timestamp"),
+					resource.TestCheckResourceAttr(dataSourceName, "closed_shards.#", acctest.Ct0),
+					resource.TestCheckResourceAttr(dataSourceName, "encryption_type", "KMS"),
+					resource.TestCheckResourceAttrPair(dataSourceName, names.AttrKMSKeyID, "aws_kms_key.test", names.AttrID),
+					resource.TestCheckResourceAttr(dataSourceName, names.AttrName, rName),
+					resource.TestCheckResourceAttr(dataSourceName, "open_shards.#", acctest.Ct2),
+					resource.TestCheckResourceAttr(dataSourceName, names.AttrStatus, "ACTIVE"),
+					resource.TestCheckResourceAttr(dataSourceName, "stream_mode_details.0.stream_mode", "PROVISIONED"),
+				),
+			},
+		},
+	})
+}
+
 func testAccStreamDataSourceConfig_basic(rName string, shardCount int) string {
 	return fmt.Sprintf(`
 resource "aws_kinesis_stream" "test" {
@@ -72,3 +101,40 @@ data "aws_kinesis_stream" "test" {
 }
 `, rName, shardCount)
 }
+func testAccStreamDataSourceConfig_encryption(rName string, shardCount int) string {
+	return fmt.Sprintf(`
+resource "aws_kinesis_stream" "test" {
+  name            = %[1]q
+  shard_count     = %[2]d
+  encryption_type = "KMS"
+  kms_key_id      = aws_kms_key.test.id
+}
+
+data "aws_kinesis_stream" "test" {
+  name = aws_kinesis_stream.test.name
+}
+
+resource "aws_kms_key" "test" {
+  description             = %[1]q
+  deletion_window_in_days = 7
+
+  policy = <<POLICY
+{
+  "Version": "2012-10-17",
+  "Id": "kms-tf-1",
+  "Statement": [
+    {
+      "Sid": "Enable IAM User Permissions",
+      "Effect": "Allow",
+      "Principal": {
+        "AWS": "*"
+      },
+      "Action": "kms:*",
+      "Resource": "*"
+    }
+  ]
+}
+POLICY
+}
+`, rName, shardCount)
+}
diff --git a/website/docs/d/kinesis_stream.html.markdown b/website/docs/d/kinesis_stream.html.markdown
index 043e4654ad4..bab2d259cd5 100644
--- a/website/docs/d/kinesis_stream.html.markdown
+++ b/website/docs/d/kinesis_stream.html.markdown
@@ -31,13 +31,15 @@ data "aws_kinesis_stream" "stream" {
 are exported:
 
 * `arn` - ARN of the Kinesis Stream (same as id).
-* `name` - Name of the Kinesis Stream.
+* `closed_shards` - List of shard ids in the CLOSED state. See [Shard State][2] for more.
 * `creation_timestamp` - Approximate UNIX timestamp that the stream was created.
-* `status` - Current status of the stream. The stream status is one of CREATING, DELETING, ACTIVE, or UPDATING.
-* `retention_period` - Length of time (in hours) data records are accessible after they are added to the stream.
+* `encryption_type` - Encryption type used.
+* `kms_key_id` - GUID for the customer-managed AWS KMS key to use for encryption.
+* `name` - Name of the Kinesis Stream.
 * `open_shards` - List of shard ids in the OPEN state. See [Shard State][2] for more.
-* `closed_shards` - List of shard ids in the CLOSED state. See [Shard State][2] for more.
+* `retention_period` - Length of time (in hours) data records are accessible after they are added to the stream.
 * `shard_level_metrics` - List of shard-level CloudWatch metrics which are enabled for the stream. See [Monitoring with CloudWatch][3] for more.
+* `status` - Current status of the stream. The stream status is one of CREATING, DELETING, ACTIVE, or UPDATING.
 * `stream_mode_details` - [Capacity mode][4] of the data stream. Detailed below.
 * `tags` - Map of tags to assigned to the stream.