Support deletion

Author: fpacifici

cmd/topicctl/subcmd/apply.go

@@ -36,6 +36,7 @@ type applyCmdConfig struct {
 	retentionDropStepDurationStr string
 	skipConfirm                  bool
 	ignoreFewerPartitionsError   bool
+	allowSettingsDeletion        bool
 	sleepLoopDuration            time.Duration
 	failFast                     bool
 
@@ -107,6 +108,12 @@ func init() {
 		false,
 		"Don't return error when topic's config specifies fewer partitions than it currently has",
 	)
+	applyCmd.Flags().BoolVar(
+		&applyConfig.allowSettingsDeletion,
+		"allow-settings-deletion",
+		false,
+		"Deletes topic settings from the broker if the setting is set on the broker but not in config",
+	)
 	applyCmd.Flags().DurationVar(
 		&applyConfig.sleepLoopDuration,
 		"sleep-loop-duration",
@@ -259,6 +266,7 @@ func applyTopic(
 			RetentionDropStepDuration:  applyConfig.retentionDropStepDuration,
 			SkipConfirm:                applyConfig.skipConfirm,
 			IgnoreFewerPartitionsError: applyConfig.ignoreFewerPartitionsError,
+			AllowSettingsDeletion:      applyConfig.allowSettingsDeletion,
 			SleepLoopDuration:          applyConfig.sleepLoopDuration,
 			TopicConfig:                topicConfig,
 		}

cmd/topicctl/subcmd/rebalance.go

@@ -307,6 +307,7 @@ func rebalanceApplyTopic(
 		AutoContinueRebalance:      true,                      // to continue without prompts
 		RetentionDropStepDuration:  retentionDropStepDuration, // not needed for rebalance
 		SkipConfirm:                true,                      // to enforce action: rebalance
+		AllowSettingsDeletion:      false,                     // Irrelevant here
 		SleepLoopDuration:          rebalanceConfig.sleepLoopDuration,
 		TopicConfig:                topicConfig,
 	}

pkg/apply/apply.go

@@ -37,6 +37,7 @@ type TopicApplierConfig struct {
 	RetentionDropStepDuration  time.Duration
 	SkipConfirm                bool
 	IgnoreFewerPartitionsError bool
+	AllowSettingsDeletion      bool
 	SleepLoopDuration          time.Duration
 	TopicConfig                config.TopicConfig
 }
@@ -392,6 +393,8 @@ func (t *TopicApplier) updateSettings(
 		return err
 	}
 
+	configEntries := []kafka.ConfigEntry{}
+
 	if len(diffKeys) > 0 {
 		diffsTable, err := FormatSettingsDiff(topicSettings, topicInfo.Config, diffKeys)
 		if err != nil {
@@ -416,6 +419,23 @@ func (t *TopicApplier) updateSettings(
 			)
 		}
 
+		configEntries, err = topicSettings.ToConfigEntries(diffKeys)
+		if err != nil {
+			return err
+		}
+	}
+
+	if len(missingKeys) > 0 && t.config.AllowSettingsDeletion {
+		log.Infof(
+			"Found %d key(s) set in cluster but missing from config for deletion:\n%s",
+			len(missingKeys),
+			FormatMissingKeys(topicInfo.Config, missingKeys),
+		)
+
+		configEntries = append(configEntries, topicSettings.ToEmptyConfigEntries(missingKeys)...)
+	}
+
+	if len(configEntries) > 0 {
 		if t.config.DryRun {
 			log.Infof("Skipping update because dryRun is set to true")
 			return nil
@@ -430,11 +450,6 @@ func (t *TopicApplier) updateSettings(
 		}
 		log.Infof("OK, updating")
 
-		configEntries, err := topicSettings.ToConfigEntries(diffKeys)
-		if err != nil {
-			return err
-		}
-
 		_, err = t.adminClient.UpdateTopicConfig(
 			ctx,
 			t.topicName,
@@ -446,7 +461,7 @@ func (t *TopicApplier) updateSettings(
 		}
 	}
 
-	if len(missingKeys) > 0 {
+	if len(missingKeys) > 0 && !t.config.AllowSettingsDeletion {
 		log.Warnf(
 			"Found %d key(s) set in cluster but missing from config:\n%s\nThese will be left as-is.",
 			len(missingKeys),

pkg/config/settings.go

@@ -346,6 +346,22 @@ func (t TopicSettings) ToConfigEntries(keys []string) ([]kafka.ConfigEntry, erro
 	return entries, nil
 }
 
+// Produces a slice of kafka-go config entries with empty value. Thus used
+// for deletion of the setting.
+func (t TopicSettings) ToEmptyConfigEntries(keys []string) []kafka.ConfigEntry {
+	entries := []kafka.ConfigEntry{}
+
+	if keys != nil {
+		for _, key := range keys {
+			entries = append(
+				entries,
+				kafka.ConfigEntry{ConfigName: key, ConfigValue: ""},
+			)
+		}
+	}
+	return entries
+}
+
 // HasKey returns whether the current settings instance contains the argument key.
 func (t TopicSettings) HasKey(key string) bool {
 	_, ok := t[key]