fix: A diff should reflect that the value of the stream arn is computed on change to the stream.

internal/service/dynamodb/table.go

@@ -74,6 +74,14 @@ func ResourceTable() *schema.Resource {
 				}
 				return nil
 			},
+			func(_ context.Context, diff *schema.ResourceDiff, v interface{}) error {
+				if diff.Id() != "" && diff.HasChange("stream_enabled") {
+					if err := diff.SetNewComputed("stream_arn"); err != nil {
+						return fmt.Errorf("error setting stream_arn to computed: %s", err)
+					}
+				}
+				return nil
+			},
 			func(_ context.Context, diff *schema.ResourceDiff, _ interface{}) error {
 				if v := diff.Get("restore_source_name"); v != "" {
 					return nil

internal/service/lambda/event_source_mapping_test.go

@@ -213,6 +213,37 @@ func TestAccLambdaEventSourceMapping_DynamoDB_functionResponseTypes(t *testing.T
 	})
 }
 
+func TestAccLambdaEventSourceMapping_DynamoDB_streamAdded(t *testing.T) {
+	var conf lambda.EventSourceMappingConfiguration
+	resourceName := "aws_dynamodb_table.test"
+	mappingResourceName := "aws_lambda_event_source_mapping.test"
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(t) },
+		ErrorCheck:               acctest.ErrorCheck(t, lambda.EndpointsID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckEventSourceMappingDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccDynamoDBStreamConfig(rName, false),
+			},
+			{
+				ResourceName:            resourceName,
+				ImportState:             true,
+				ImportStateVerify:       true,
+				ImportStateVerifyIgnore: []string{"last_modified"},
+			},
+			{
+				Config: testAccEventSourceMappingDynamoDBStreamEnabled(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckEventSourceMappingExists(mappingResourceName, &conf),
+				),
+			},
+		},
+	})
+}
+
 func TestAccLambdaEventSourceMapping_SQS_batchWindow(t *testing.T) {
 	var conf lambda.EventSourceMappingConfiguration
 	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
@@ -1326,6 +1357,78 @@ resource "aws_lambda_function" "test" {
 `, rName)
 }
 
+func testAccDynamoDBStreamConfig(rName string, streamEnabled bool) string {
+	var streamStatus string
+	var streamViewType string
+	if streamEnabled {
+		streamStatus = "stream_enabled   = true"
+		streamViewType = "stream_view_type = \"KEYS_ONLY\""
+	} else {
+		streamStatus = ""
+		streamViewType = ""
+	}
+
+	return fmt.Sprintf(`
+resource "aws_iam_role" "test" {
+  name = %[1]q
+
+  assume_role_policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [{
+    "Action": "sts:AssumeRole",
+    "Principal": {
+      "Service": "lambda.amazonaws.com"
+    },
+    "Effect": "Allow"
+  }]
+}
+EOF
+}
+
+resource "aws_iam_role_policy" "test" {
+  role = aws_iam_role.test.name
+
+  policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": [
+        "dynamodb:*"
+      ],
+      "Resource": "*"
+    }
+  ]
+}
+EOF
+}
+
+resource "aws_dynamodb_table" "test" {
+  name           = %[1]q
+  read_capacity  = 1
+  write_capacity = 2
+  hash_key       = "TestTableHashKey"
+
+  attribute {
+    name = "TestTableHashKey"
+    type = "S"
+  }
+  %[2]s
+  %[3]s
+}
+
+resource "aws_lambda_function" "test" {
+  filename      = "test-fixtures/lambdatest.zip"
+  function_name = %[1]q
+  handler       = "exports.example"
+  role          = aws_iam_role.test.arn
+  runtime       = "nodejs12.x"
+}
+`, rName, streamStatus, streamViewType)
+}
+
 func testAccEventSourceMappingConfig_kafkaBase(rName string) string {
 	return acctest.ConfigCompose(acctest.ConfigVPCWithSubnets(rName, 2), fmt.Sprintf(`
 resource "aws_iam_role" "test" {
@@ -1967,6 +2070,18 @@ resource "aws_lambda_event_source_mapping" "test" {
 `)
 }
 
+func testAccEventSourceMappingDynamoDBStreamEnabled(rName string) string {
+	return acctest.ConfigCompose(testAccDynamoDBStreamConfig(rName, true), `
+resource "aws_lambda_event_source_mapping" "test" {
+  batch_size        = 150
+  enabled           = true
+  event_source_arn  = aws_dynamodb_table.test.stream_arn
+  function_name     = aws_lambda_function.test.function_name
+  starting_position = "LATEST"
+}
+`)
+}
+
 func testAccEventSourceMappingConfig_activeMQ(rName, batchSize string) string {
 	if batchSize == "" {
 		batchSize = "null"