Update Stream.yaml to add cdc method support for MySQL (#12465)

Author: ravijaingit

mmv1/products/datastream/Stream.yaml

@@ -133,6 +133,22 @@ examples:
       'deletion_protection': 'false'
     # Requires SQLServer Configuration
     exclude_test: true
+  - name: 'datastream_stream_mysql_gtid'
+    primary_resource_id: 'default'
+    vars:
+      database_name: 'db'
+      database_password: 'password'
+      database_user: 'user'
+      deletion_protection: 'true'
+      destination_connection_profile_id: 'destination-profile'
+      source_connection_profile_id: 'source-profile'
+      mysql_name: 'mysql'
+      mysql_root_password: 'root-password'
+      stream_id: 'stream'
+    test_vars_overrides:
+      'deletion_protection': 'false'
+    # Requires MySQL Configuration
+    exclude_test: true
   - name: 'datastream_stream_postgresql_bigquery_dataset_id'
     primary_resource_id: 'default'
     vars:
@@ -420,6 +436,24 @@ properties:
             send_empty_value: true
             validation:
               function: 'validation.IntAtLeast(0)'
+          - name: 'binaryLogPosition'
+            type: NestedObject
+            allow_empty_object: true
+            send_empty_value: true
+            conflicts:
+              - source_config.0.mysql_source_config.gtid
+            description: |
+              CDC reader reads from binary logs replication cdc method.
+            properties: []
+          - name: 'gtid'
+            type: NestedObject
+            allow_empty_object: true
+            send_empty_value: true
+            conflicts:
+              - source_config.0.mysql_source_config.binaryLogPosition
+            description: |
+              CDC reader reads from gtid based replication.
+            properties: []
       - name: 'oracleSourceConfig'
         type: NestedObject
         description: |

mmv1/templates/terraform/examples/datastream_stream_mysql_gtid.tf.tmpl

@@ -0,0 +1,103 @@
+resource "google_sql_database_instance" "instance" {
+    name                = "<%= ctx[:vars]['mysql_name'] %>"
+    database_version    = "MYSQL_8_0"
+    region              = "us-central1"
+    root_password       = "<%= ctx[:vars]['mysql_root_password'] %>"
+    deletion_protection = "<%= ctx[:vars]['deletion_protection'] %>"
+
+    settings {
+        tier = "db-custom-2-4096"
+        ip_configuration {
+            // Datastream IPs will vary by region.
+            // https://cloud.google.com/datastream/docs/ip-allowlists-and-regions
+            authorized_networks {
+                value = "34.71.242.81"
+            }
+
+            authorized_networks {
+                value = "34.72.28.29"
+            }
+
+            authorized_networks {
+                value = "34.67.6.157"
+            }
+
+            authorized_networks {
+                value = "34.67.234.134"
+            }
+
+            authorized_networks {
+                value = "34.72.239.218"
+            }
+        }
+    }
+}
+
+resource "google_sql_database" "db" {
+    name       = "<%= ctx[:vars]['database_name'] %>"
+    instance   = google_sql_database_instance.instance.name
+    depends_on = [google_sql_user.user]
+}
+
+resource "google_sql_user" "user" {
+    name     = "<%= ctx[:vars]['database_user'] %>"
+    instance = google_sql_database_instance.instance.name
+    password = "<%= ctx[:vars]['database_password'] %>"
+}
+
+resource "google_datastream_connection_profile" "source" {
+    display_name          = "MySQL Source"
+    location              = "us-central1"
+    connection_profile_id = "<%= ctx[:vars]['source_connection_profile_id'] %>"
+
+    mysql_profile {
+        hostname = google_sql_database_instance.instance.public_ip_address
+        port     = 1433
+        username = google_sql_user.user.name
+        password = google_sql_user.user.password
+        database = google_sql_database.db.name
+    }
+}
+
+resource "google_datastream_connection_profile" "destination" {
+    display_name          = "BigQuery Destination"
+    location              = "us-central1"
+    connection_profile_id = "<%= ctx[:vars]['destination_connection_profile_id'] %>"
+
+    bigquery_profile {}
+}
+
+resource "google_datastream_stream" "default" {
+    display_name = "MySQL to BigQuery"
+    location     = "us-central1"
+    stream_id    = "<%= ctx[:vars]['stream_id'] %>"
+
+    source_config {
+        source_connection_profile = google_datastream_connection_profile.source.id
+        mysql_source_config {
+            include_objects {
+                schemas {
+                    schema = "schema"
+                    tables {
+                        table = "table"
+                    }
+                }
+            }
+            gtid {}
+        }
+    }
+
+    destination_config {
+        destination_connection_profile = google_datastream_connection_profile.destination.id
+        bigquery_destination_config {
+            data_freshness = "900s"
+            source_hierarchy_datasets {
+                dataset_template {
+                    location = "us-central1"
+                }
+            }
+        }
+    }
+
+    backfill_none {}
+}

mmv1/third_party/terraform/services/datastream/resource_datastream_stream_test.go

@@ -65,10 +65,20 @@ func TestAccDatastreamStream_update(t *testing.T) {
 				ImportStateVerify:       true,
 				ImportStateVerifyIgnore: []string{"create_without_validation", "stream_id", "location", "desired_state", "labels", "terraform_labels"},
 			},
+			{
+				ResourceName:            "google_datastream_stream.gtid",
+				ImportState:             true,
+				ImportStateVerify:       true,
+				ImportStateVerifyIgnore: []string{"create_without_validation", "stream_id", "location", "desired_state", "labels", "terraform_labels"},
+			},
 			{
 				// Disable prevent_destroy
 				Config: testAccDatastreamStream_datastreamStreamBasicUpdate(context, "RUNNING", false),
 			},
+			{
+				Config: testAccDatastreamStream_datastreamStreamBasicExample(context),
+				Check:  resource.TestCheckResourceAttr("google_datastream_stream.gtid", "state", "NOT_STARTED"),
+			},
 		},
 	})
 }
@@ -204,7 +214,45 @@ resource "google_datastream_stream" "default" {
     source_config {
         source_connection_profile = google_datastream_connection_profile.source_connection_profile.id
 
-        mysql_source_config {}
+        mysql_source_config {
+	        binary_log_position {}
+	    }
+    }
+    destination_config {
+        destination_connection_profile = google_datastream_connection_profile.destination_connection_profile.id
+        gcs_destination_config {
+            path = "mydata"
+            file_rotation_mb = 200
+            file_rotation_interval = "60s"
+            json_file_format {
+                schema_file_format = "NO_SCHEMA_FILE"
+                compression = "GZIP"
+            }
+        }
+    }
+
+    backfill_all {
+    }
+	%{lifecycle_block}
+}
+
+resource "google_datastream_stream" "gtid" {
+    stream_id = "tf-test-my-stream-gtid%{random_suffix}"
+    location = "us-central1"
+    display_name = "my gtid stream update"
+    desired_state = "%{desired_state}"
+    create_without_validation = true
+
+    labels = {
+    	key = "updated"
+    }
+
+    source_config {
+        source_connection_profile = google_datastream_connection_profile.source_connection_profile.id
+
+        mysql_source_config {
+	        gtid {}
+	    }
     }
     destination_config {
         destination_connection_profile = google_datastream_connection_profile.destination_connection_profile.id