feat(kafka topic create): add cleanup policy flag

pkg/cmd/kafka/topic/create/create.go

@@ -32,6 +32,7 @@ import (
 const (
 	defaultRetentionPeriodMS = 604800000
 	defaultRetentionSize     = -1
+	defaultCleanupPolicy     = "delete"
 )
 
 type Options struct {
@@ -41,6 +42,7 @@ type Options struct {
 	retentionBytes int
 	kafkaID        string
 	outputFormat   string
+	cleanupPolicy  string
 	interactive    bool
 
 	IO         *iostreams.IOStreams
@@ -77,6 +79,12 @@ func NewCreateTopicCommand(f *factory.Factory) *cobra.Command {
 				return err
 			}
 
+			// check that a valid --cleanup-policy flag value is used
+			validPolicy := flagutil.IsValidInput(opts.cleanupPolicy, topicutil.ValidCleanupPolicies...)
+			if !validPolicy {
+				return flag.InvalidValueError("cleanup-policy", opts.cleanupPolicy, topicutil.ValidCleanupPolicies...)
+			}
+
 			if !opts.interactive {
 
 				validator := &topicutil.Validator{
@@ -125,9 +133,12 @@ func NewCreateTopicCommand(f *factory.Factory) *cobra.Command {
 	cmd.Flags().Int32Var(&opts.partitions, "partitions", 1, opts.localizer.MustLocalize("kafka.topic.common.input.partitions.description"))
 	cmd.Flags().IntVar(&opts.retentionMs, "retention-ms", defaultRetentionPeriodMS, opts.localizer.MustLocalize("kafka.topic.common.input.retentionMs.description"))
 	cmd.Flags().IntVar(&opts.retentionBytes, "retention-bytes", defaultRetentionSize, opts.localizer.MustLocalize("kafka.topic.common.input.retentionBytes.description"))
+	cmd.Flags().StringVar(&opts.cleanupPolicy, "cleanup-policy", defaultCleanupPolicy, opts.localizer.MustLocalize("kafka.topic.common.input.cleanupPolicy.description"))
 
 	flagutil.EnableOutputFlagCompletion(cmd)
 
+	flagutil.EnableStaticFlagCompletion(cmd, "cleanup-policy", topicutil.ValidCleanupPolicies)
+
 	return cmd
 }
 
@@ -271,15 +282,29 @@ func runInteractivePrompt(opts *Options) (err error) {
 		return err
 	}
 
+	cleanupPolicyPrompt := &survey.Select{
+		Message: opts.localizer.MustLocalize("kafka.topic.create.input.cleanupPolicy.message"),
+		Help:    opts.localizer.MustLocalize("kafka.topic.common.input.cleanupPolicy.description"),
+		Options: topicutil.ValidCleanupPolicies,
+		Default: defaultCleanupPolicy,
+	}
+
+	err = survey.AskOne(cleanupPolicyPrompt, &opts.cleanupPolicy)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
 func createConfigEntries(opts *Options) *[]kafkainstanceclient.ConfigEntry {
 	retentionMsStr := strconv.Itoa(opts.retentionMs)
 	retentionBytesStr := strconv.Itoa(opts.retentionBytes)
+	cleanupPolicyStr := opts.cleanupPolicy
 	configEntryMap := map[string]*string{
 		topicutil.RetentionMsKey:   &retentionMsStr,
 		topicutil.RetentionSizeKey: &retentionBytesStr,
+		topicutil.CleanupPolicy:    &cleanupPolicyStr,
 	}
 	return topicutil.CreateConfigEntries(configEntryMap)
 }

pkg/kafka/topic/util.go

@@ -9,6 +9,10 @@ import (
 
 var RetentionMsKey string = "retention.ms"
 var RetentionSizeKey string = "retention.bytes"
+var PartitionsKey string = "partitions"
+var CleanupPolicy string = "cleanup.policy"
+
+var ValidCleanupPolicies = []string{"delete", "compact", "compact, delete"}
 
 // CreateConfigEntries converts a key value map of config entries to an array of config entries
 func CreateConfigEntries(entryMap map[string]*string) *[]kafkainstanceclient.ConfigEntry {

pkg/localize/locales/en/cmd/kafka_topic_common.en.toml

@@ -16,6 +16,10 @@ one = 'The period of time in milliseconds the broker will retain a partition log
 description = 'Description for the Retention size input'
 one = 'The maximum total size of a partition log segments before old log segments are deleted to free up space'
 
+[kafka.topic.common.input.cleanupPolicy.description]
+description = 'Description for the Cleanup policy input'
+one = 'Determines whether log messages are deleted, compacted, or both'
+
 [kafka.topic.common.error.noKafkaSelected]
 one = 'no Kafka instance is currently selected, run "rhoas kafka use" to set the current instance'
 

pkg/localize/locales/en/cmd/kafka_topic_create.en.toml

@@ -37,4 +37,8 @@ one = 'Retention Size (bytes):'
 
 [kafka.topic.create.input.partitions.message]
 description = 'Message for the Partitions input'
-one = 'Number of Partitions:'
+one = 'Number of Partitions:'
+
+[kafka.topic.create.input.cleanupPolicy.message]
+description = 'Message for the Cleanup Policy input'
+one = 'Cleanup Policy:'