feat: Add Pub/Sub Topic support for changing schema settings (#8078) (#14819)

* Add support for Pub/Sub schema evolution

* Add example for Pub/Sub schema evolution

* Undo changes to Schema

* Remove unneeded Topic changes.

* Remove accidentally added newline

* Added test for schema update

Signed-off-by: Modular Magician <[email protected]>

Author: modular-magician

.changelog/8078.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+pubsub: Allowed `schema_settings` of `google_pubsub_topic` to change without deleting and recreating the resource
+```

google/resource_pubsub_topic.go

@@ -119,7 +119,6 @@ and is not a valid configuration.`,
 						"schema": {
 							Type:     schema.TypeString,
 							Required: true,
-							ForceNew: true,
 							Description: `The name of the schema that messages published should be
 validated against. Format is projects/{project}/schemas/{schema}.
 The value of this field will be _deleted-schema_
@@ -128,7 +127,6 @@ if the schema has been deleted.`,
 						"encoding": {
 							Type:         schema.TypeString,
 							Optional:     true,
-							ForceNew:     true,
 							ValidateFunc: verify.ValidateEnum([]string{"ENCODING_UNSPECIFIED", "JSON", "BINARY", ""}),
 							Description:  `The encoding of messages validated against schema. Default value: "ENCODING_UNSPECIFIED" Possible values: ["ENCODING_UNSPECIFIED", "JSON", "BINARY"]`,
 							Default:      "ENCODING_UNSPECIFIED",

google/resource_pubsub_topic_test.go

@@ -69,6 +69,40 @@ func TestAccPubsubTopic_cmek(t *testing.T) {
 	})
 }
 
+func TestAccPubsubTopic_schema(t *testing.T) {
+	t.Parallel()
+
+	schema1 := fmt.Sprintf("tf-test-schema-%s", RandString(t, 10))
+	schema2 := fmt.Sprintf("tf-test-schema-%s", RandString(t, 10))
+	topic := fmt.Sprintf("tf-test-topic-%s", RandString(t, 10))
+
+	VcrTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.AccTestPreCheck(t) },
+		ProtoV5ProviderFactories: ProtoV5ProviderFactories(t),
+		CheckDestroy:             testAccCheckPubsubTopicDestroyProducer(t),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccPubsubTopic_updateWithSchema(topic, schema1),
+			},
+			{
+				ResourceName:      "google_pubsub_topic.bar",
+				ImportStateId:     topic,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccPubsubTopic_updateWithNewSchema(topic, schema2),
+			},
+			{
+				ResourceName:      "google_pubsub_topic.bar",
+				ImportStateId:     topic,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func testAccPubsubTopic_update(topic, key, value string) string {
 	return fmt.Sprintf(`
 resource "google_pubsub_topic" "foo" {
@@ -105,3 +139,39 @@ resource "google_pubsub_topic" "topic" {
 }
 `, topicName, kmsKey)
 }
+
+func testAccPubsubTopic_updateWithSchema(topic, schema string) string {
+	return fmt.Sprintf(`
+resource "google_pubsub_schema" "foo" {
+	name = "%s"
+	type = "PROTOCOL_BUFFER"
+  definition = "syntax = \"proto3\";\nmessage Results {\nstring f1 = 1;\n}"
+}
+
+resource "google_pubsub_topic" "bar" {
+  name = "%s"
+	schema_settings {
+    schema = google_pubsub_schema.foo.id
+    encoding = "BINARY"
+  }
+}
+`, schema, topic)
+}
+
+func testAccPubsubTopic_updateWithNewSchema(topic, schema string) string {
+	return fmt.Sprintf(`
+resource "google_pubsub_schema" "foo" {
+	name = "%s"
+	type = "PROTOCOL_BUFFER"
+	definition = "syntax = \"proto3\";\nmessage Results {\nstring f1 = 1;\n}"
+}
+
+resource "google_pubsub_topic" "bar" {
+  name = "%s"
+	schema_settings {
+    schema = google_pubsub_schema.foo.id
+    encoding = "JSON"
+  }
+}
+`, schema, topic)
+}