Merge pull request #75 from redpanda-data/topic-cfg-remove

topic config: support removal of state configs

Author: r-vasquez

examples/datasource/main.tf

@@ -7,6 +7,15 @@ data "redpanda_cluster" "test" {
   id = var.cluster_id
 }
 
+resource "redpanda_topic" "test" {
+  name               = var.topic_name
+  partition_count    = var.partition_count
+  replication_factor = var.replication_factor
+  cluster_api_url    = data.redpanda_cluster.test.cluster_api_url
+  allow_deletion     = true
+  configuration      = var.topic_config
+}
+
 resource "redpanda_user" "test" {
   name            = var.user_name
   password        = var.user_pw
@@ -25,6 +34,12 @@ resource "redpanda_acl" "test" {
   cluster_api_url       = data.redpanda_cluster.test.cluster_api_url
 }
 
+variable "topic_config" {
+    default = {
+      "cleanup.policy"   = "compact"
+      "compression.type" = "snappy"
+    }
+}
 variable "user_name" {
   default = "test-username"
 }

redpanda/resources/topic/resource_topic.go

@@ -256,12 +256,12 @@ func (t *Topic) Update(ctx context.Context, request resource.UpdateRequest, resp
 	}
 	defer t.dataplaneConn.Close()
 	if !plan.Configuration.Equal(state.Configuration) {
-		cfgToSet, err := utils.MapToUpdateTopicConfiguration(plan.Configuration)
+		cfgToSet, err := parseUpdateConfiguration(state, plan)
 		if err != nil {
 			response.Diagnostics.AddError("failed to parse the configuration map", err.Error())
 			return
 		}
-		cfgs, err := t.TopicClient.UpdateTopicConfigurations(ctx, &dataplanev1alpha1.UpdateTopicConfigurationsRequest{
+		cfgs, err := t.TopicClient.SetTopicConfigurations(ctx, &dataplanev1alpha1.SetTopicConfigurationsRequest{
 			TopicName:      plan.Name.ValueString(),
 			Configurations: cfgToSet,
 		})
@@ -361,3 +361,41 @@ func filterDynamicConfig(configs []*dataplanev1alpha1.Topic_Configuration) []*da
 func isAlreadyExistsError(err error) bool {
 	return strings.Contains(err.Error(), "TOPIC_ALREADY_EXISTS") || strings.Contains(err.Error(), "The topic has already been created")
 }
+
+// parseUpdateConfiguration parses both the state and the plan configuration,
+// and generates a new SetConfiguration map with:
+//   - Configurations that are both in the state, and the plan.
+//   - Configurations that are in the plan and not in the state.
+//   - Updated configurations that are in the plan.
+func parseUpdateConfiguration(state, plan models.Topic) ([]*dataplanev1alpha1.SetTopicConfigurationsRequest_SetConfiguration, error) {
+	var output []*dataplanev1alpha1.SetTopicConfigurationsRequest_SetConfiguration
+	// Loop through state configuration.
+	for k, v := range state.Configuration.Elements() {
+		if v.IsNull() || v.IsUnknown() {
+			return nil, fmt.Errorf("topic configuration %q must have a value", k)
+		}
+		// If configuration also exists in plan.
+		if pv, ok := plan.Configuration.Elements()[k]; ok {
+			// Add whatever is in the plan to the output.
+			value := strings.Trim(pv.String(), `"`)
+			output = append(output, &dataplanev1alpha1.SetTopicConfigurationsRequest_SetConfiguration{
+				Name:  k,
+				Value: &value,
+			})
+		}
+		// If it's in the state, but not in the plan, then the
+		// config was removed. We do not add that.
+	}
+	// Now we check for what's in the plan and not in the state.
+	for k, v := range plan.Configuration.Elements() {
+		// Add to output if it's not in the state.
+		if _, ok := state.Configuration.Elements()[k]; !ok {
+			value := strings.Trim(v.String(), `"`)
+			output = append(output, &dataplanev1alpha1.SetTopicConfigurationsRequest_SetConfiguration{
+				Name:  k,
+				Value: &value,
+			})
+		}
+	}
+	return output, nil
+}

redpanda/tests/acceptance_test.go

@@ -28,6 +28,7 @@ const (
 	networkResourceName   = "redpanda_network.test"
 	clusterResourceName   = "redpanda_cluster.test"
 	userResourceName      = "redpanda_user.test"
+	topicResourceName     = "redpanda_topic.test"
 	aclResourceName       = "redpanda_acl.test"
 )
 
@@ -604,6 +605,14 @@ func TestAccResourcesWithDataSources(t *testing.T) {
 	maps.Copy(origTestCaseVars, providerCfgIDSecretVars)
 	origTestCaseVars["cluster_id"] = config.StringVariable(os.Getenv("CLUSTER_ID"))
 	origTestCaseVars["user_name"] = config.StringVariable(name)
+	origTestCaseVars["topic_name"] = config.StringVariable(name)
+
+	updateTestCaseVars := make(map[string]config.Variable)
+	maps.Copy(updateTestCaseVars, origTestCaseVars)
+	// Change 1, remove other
+	updateTestCaseVars["topic_config"] = config.MapVariable(map[string]config.Variable{
+		"compression.type": config.StringVariable("gzip"),
+	})
 
 	c, err := newClients(ctx, clientID, clientSecret, "ign")
 	if err != nil {
@@ -618,6 +627,15 @@ func TestAccResourcesWithDataSources(t *testing.T) {
 				ProtoV6ProviderFactories: testAccProtoV6ProviderFactories,
 				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttr(userResourceName, "name", name),
+					resource.TestCheckResourceAttr(topicResourceName, "name", name),
+				),
+			},
+			{
+				ConfigFile:               config.StaticFile(dataSourcesTest),
+				ConfigVariables:          updateTestCaseVars,
+				ProtoV6ProviderFactories: testAccProtoV6ProviderFactories,
+				Check: resource.ComposeAggregateTestCheckFunc(
+					resource.TestCheckResourceAttr(topicResourceName, "configuration.compression.type", "gzip"),
 				),
 			},
 			{

redpanda/utils/utils.go

@@ -335,24 +335,6 @@ func MapToCreateTopicConfiguration(cfg types.Map) ([]*dataplanev1alpha1.CreateTo
 	return output, nil
 }
 
-// MapToUpdateTopicConfiguration converts a cfg map to a slice of dataplanev1alpha1.UpdateTopicConfigurationsRequest_UpdateConfiguration
-func MapToUpdateTopicConfiguration(cfg types.Map) ([]*dataplanev1alpha1.UpdateTopicConfigurationsRequest_UpdateConfiguration, error) {
-	var output []*dataplanev1alpha1.UpdateTopicConfigurationsRequest_UpdateConfiguration
-
-	for k, v := range cfg.Elements() {
-		if v.IsNull() || v.IsUnknown() {
-			return nil, fmt.Errorf("topic configuration %q must have a value", k)
-		}
-		value := strings.Trim(v.String(), `"`)
-		output = append(output, &dataplanev1alpha1.UpdateTopicConfigurationsRequest_UpdateConfiguration{
-			Name:      k,
-			Value:     &value,
-			Operation: dataplanev1alpha1.ConfigAlterOperation_CONFIG_ALTER_OPERATION_SET,
-		})
-	}
-	return output, nil
-}
-
 // NumberToInt32 converts a types.Number to an *int32
 func NumberToInt32(n types.Number) *int32 {
 	i, _ := n.ValueBigFloat().Int64()