INC-3971 Fix TF intermittent false diff detection issue

internal/provider/resource_kafka_topic.go

@@ -118,6 +118,27 @@ func kafkaTopicResource() *schema.Resource {
 				Optional:    true,
 				Computed:    true,
 				Description: "The custom topic settings to set (e.g., `\"cleanup.policy\" = \"compact\"`).",
+				DiffSuppressFunc: func(k, old, new string, d *schema.ResourceData) bool {
+					if old == "" && new == "" {
+						return true
+					}
+
+					// when old value exists but new value is empty, check if key is in new config and suppress diff if not 
+					if new == "" && old != "" {
+						configKey := k
+						if strings.HasPrefix(k, "config.") {
+							configKey = strings.TrimPrefix(k, "config.")
+						}
+
+						if newConfigs, ok := d.GetOk(paramConfigs); ok {
+							newConfigMap := newConfigs.(map[string]interface{})
+							_, keyExists := newConfigMap[configKey]
+
+							return !keyExists
+						}
+					}
+					return false
+				},
 			},
 			paramCredentials: credentialsSchema(),
 		},

internal/provider/resource_kafka_topic_test.go

@@ -26,6 +26,7 @@
 
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 )
 
 const (
@@ -200,7 +201,7 @@
 					resource.TestCheckResourceAttr(fullTopicResourceLabel, "rest_endpoint", mockTopicTestServerUrl),
 					resource.TestCheckResourceAttr(fullTopicResourceLabel, "config.%", "2"),
 					resource.TestCheckResourceAttr(fullTopicResourceLabel, "config.max.message.bytes", "12345"),
 					resource.TestCheckResourceAttr(fullTopicResourceLabel, "config.retention.ms", "6789"),
 					resource.TestCheckNoResourceAttr(fullTopicResourceLabel, fmt.Sprintf("config.%s", fifthConfigName)),
 					resource.TestCheckResourceAttr(fullTopicResourceLabel, "credentials.#", "1"),
 					resource.TestCheckResourceAttr(fullTopicResourceLabel, "credentials.0.%", "2"),
@@ -503,6 +504,36 @@
 	checkStubCount(t, wiremockClient, deleteTopicStub, fmt.Sprintf("DELETE %s", kafkaTopicPath), expectedCountTwo)
 }
 
+func TestAccTopicConfigDiffSuppression(t *testing.T) {
+	d := schema.TestResourceDataRaw(t, kafkaTopicResource().Schema, map[string]interface{}{
+		"config": map[string]interface{}{
+			"retention.ms": "604800000",
+		},
+	})
+
+	diffSuppressFunc := kafkaTopicResource().Schema["config"].DiffSuppressFunc
+
+	result1 := diffSuppressFunc("config.cleanup.policy", "", "", d)
+	if !result1 {
+		t.Error("Expected diff to be suppressed when both old and new values are empty")
+	}
+
+	result2 := diffSuppressFunc("config.cleanup.policy", "delete", "", d)
+	if !result2 {
+		t.Error("Expected diff to be suppressed when old value exists but key is not in new config")
+	}
+
+	result3 := diffSuppressFunc("config.retention.ms", "123456", "", d)
+	if result3 {
+		t.Error("Expected diff NOT to be suppressed when key exists in new config but values differ")
+	}
+
+	result4 := diffSuppressFunc("config.retention.ms", "123456", "604800000", d)
+	if result4 {
+		t.Error("Expected diff NOT to be suppressed when both values exist but are different")
+	}
+}
+
 func testAccCheckTopicDestroy(s *terraform.State, url string) error {
 	testClient := testAccProvider.Meta().(*Client)
 	c := testClient.kafkaRestClientFactory.CreateKafkaRestClient(url, clusterId, kafkaApiKey, kafkaApiSecret, false, false, testClient.oauthToken)