Merge pull request #41794 from sappusaketh/f-aws_kinesis_firehose_delivery_stream-msk_source_configuration-read_from_timestamp

feat: add read_from_timestamp option to msk_source_configuration for Firehose delivery stream

Author: ewbankkit

.changelog/41794.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_kinesis_firehose_delivery_stream: Add `msk_source_configuration.read_from_timestamp` argument
+```

internal/service/firehose/delivery_stream.go

@@ -957,6 +957,12 @@ func resourceDeliveryStream() *schema.Resource {
 								ForceNew:     true,
 								ValidateFunc: verify.ValidARN,
 							},
+							"read_from_timestamp": {
+								Type:         schema.TypeString,
+								Optional:     true,
+								ForceNew:     true,
+								ValidateFunc: validation.IsRFC3339Time,
+							},
 							"topic_name": {
 								Type:     schema.TypeString,
 								Required: true,
@@ -3493,6 +3499,11 @@ func expandMSKSourceConfiguration(tfMap map[string]interface{}) *types.MSKSource
 		apiObject.MSKClusterARN = aws.String(v)
 	}
 
+	if v, ok := tfMap["read_from_timestamp"].(string); ok && v != "" {
+		v, _ := time.Parse(time.RFC3339, v)
+		apiObject.ReadFromTimestamp = aws.Time(v)
+	}
+
 	if v, ok := tfMap["topic_name"].(string); ok && v != "" {
 		apiObject.TopicName = aws.String(v)
 	}
@@ -3533,6 +3544,10 @@ func flattenMSKSourceDescription(apiObject *types.MSKSourceDescription) map[stri
 		tfMap["msk_cluster_arn"] = aws.ToString(v)
 	}
 
+	if v := apiObject.ReadFromTimestamp; v != nil {
+		tfMap["read_from_timestamp"] = aws.ToTime(v).Format(time.RFC3339)
+	}
+
 	if v := apiObject.TopicName; v != nil {
 		tfMap["topic_name"] = aws.ToString(v)
 	}

internal/service/firehose/delivery_stream_test.go

@@ -24,6 +24,16 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/names"
 )
 
+func init() {
+	acctest.RegisterServiceErrorCheckFunc(names.FirehoseServiceID, testAccErrorCheckSkip)
+}
+
+func testAccErrorCheckSkip(t *testing.T) resource.ErrorCheckFunc {
+	return acctest.ErrorCheckSkipMessagesContaining(t,
+		"Read from timestamp feature is not currently available",
+	)
+}
+
 func TestAccFirehoseDeliveryStream_basic(t *testing.T) {
 	ctx := acctest.Context(t)
 	var stream types.DeliveryStreamDescription
@@ -996,6 +1006,42 @@ func TestAccFirehoseDeliveryStream_ExtendedS3_mskClusterSource(t *testing.T) {
 					resource.TestCheckResourceAttrSet(resourceName, "msk_source_configuration.0.authentication_configuration.0.role_arn"),
 					resource.TestCheckResourceAttrSet(resourceName, "msk_source_configuration.0.msk_cluster_arn"),
 					resource.TestCheckResourceAttr(resourceName, "msk_source_configuration.0.topic_name", "test"),
+					resource.TestCheckResourceAttr(resourceName, "msk_source_configuration.0.read_from_timestamp", ""),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func TestAccFirehoseDeliveryStream_ExtendedS3_readFromTimestamp(t *testing.T) {
+	ctx := acctest.Context(t)
+	var stream types.DeliveryStreamDescription
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_kinesis_firehose_delivery_stream.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.FirehoseServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckDeliveryStreamDestroy(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDeliveryStreamConfig_readFromTimestamp(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckDeliveryStreamExists(ctx, resourceName, &stream),
+					resource.TestCheckResourceAttr(resourceName, "kinesis_source_configuration.#", "0"),
+					resource.TestCheckResourceAttr(resourceName, "msk_source_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "msk_source_configuration.0.authentication_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "msk_source_configuration.0.authentication_configuration.0.connectivity", "PRIVATE"),
+					resource.TestCheckResourceAttrSet(resourceName, "msk_source_configuration.0.authentication_configuration.0.role_arn"),
+					resource.TestCheckResourceAttrSet(resourceName, "msk_source_configuration.0.msk_cluster_arn"),
+					resource.TestCheckResourceAttr(resourceName, "msk_source_configuration.0.topic_name", "test"),
+					resource.TestCheckResourceAttr(resourceName, "msk_source_configuration.0.read_from_timestamp", "2025-03-11T14:30:00Z"),
 				),
 			},
 			{
@@ -3303,6 +3349,36 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 `, rName))
 }
 
+func testAccDeliveryStreamConfig_readFromTimestamp(rName string) string {
+	return acctest.ConfigCompose(
+		testAccDeliveryStreamConfig_base(rName),
+		testAccDeliveryStreamConfig_baseMSKClusterSource(rName),
+		fmt.Sprintf(`
+resource "aws_kinesis_firehose_delivery_stream" "test" {
+  depends_on = [aws_iam_role_policy.firehose, aws_iam_role_policy.msk_source, aws_msk_cluster_policy.test]
+  name       = %[1]q
+
+  msk_source_configuration {
+    authentication_configuration {
+      connectivity = "PRIVATE"
+      role_arn     = aws_iam_role.msk_source.arn
+    }
+
+    msk_cluster_arn     = aws_msk_serverless_cluster.test.arn
+    topic_name          = "test"
+    read_from_timestamp = "2025-03-11T14:30:00Z"
+  }
+
+  destination = "extended_s3"
+
+  extended_s3_configuration {
+    role_arn   = aws_iam_role.firehose.arn
+    bucket_arn = aws_s3_bucket.bucket.arn
+  }
+}
+`, rName))
+}
+
 func testAccDeliveryStreamConfig_extendedS3DataFormatConversionConfigurationEnabled(rName string, enabled bool) string {
 	return acctest.ConfigCompose(testAccDeliveryStreamConfig_base(rName), fmt.Sprintf(`
 resource "aws_glue_catalog_database" "test" {

website/docs/r/kinesis_firehose_delivery_stream.html.markdown

@@ -735,6 +735,7 @@ The `msk_source_configuration` configuration block supports the following argume
 * `authentication_configuration` - (Required) The authentication configuration of the Amazon MSK cluster. See [`authentication_configuration` block](#authentication_configuration-block) below for details.
 * `msk_cluster_arn` - (Required) The ARN of the Amazon MSK cluster.
 * `topic_name` - (Required) The topic name within the Amazon MSK cluster.
+* `read_from_timestamp` - (Optional) The start date and time in UTC for the offset position within your MSK topic from where Firehose begins to read. By default, this is set to timestamp when Firehose becomes Active. If you want to create a Firehose stream with Earliest start position set the `read_from_timestamp` parameter to Epoch (1970-01-01T00:00:00Z).
 
 ### `authentication_configuration` block