[CLI-3530] Only display produce/consume info messages when the verbosity level is at least info (#3208)

Author: sgagniere

internal/asyncapi/command_export.go

@@ -708,7 +708,7 @@ func addComponents(reflector asyncapi.Reflector, messages map[string]spec.Messag
 }
 
 func createConsumer(broker string, clusterCreds *config.APIKeyPair, group string) (*ckgo.Consumer, error) {
-	consumer, err := ckgo.NewConsumer(&ckgo.ConfigMap{
+	configMap := &ckgo.ConfigMap{
 		"bootstrap.servers":  broker,
 		"sasl.mechanisms":    "PLAIN",
 		"security.protocol":  "SASL_SSL",
@@ -717,7 +717,11 @@ func createConsumer(broker string, clusterCreds *config.APIKeyPair, group string
 		"group.id":           group,
 		"auto.offset.reset":  "earliest",
 		"enable.auto.commit": "false",
-	})
+	}
+	if err := kafka.SetConsumerDebugOption(configMap); err != nil {
+		return nil, fmt.Errorf("failed to create Kafka consumer: %w", err)
+	}
+	consumer, err := ckgo.NewConsumer(configMap)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create Kafka consumer: %w", err)
 	}

internal/kafka/confluent_kafka_configs.go

@@ -334,23 +334,33 @@ func getPartitionsByIndex(partitions []ckgo.TopicPartition, partitionFilter Part
 }
 
 func SetProducerDebugOption(configMap *ckgo.ConfigMap) error {
+	// Note: log_levels are based on syslog levels
 	switch log.CliLogger.Level {
+	case log.WARN:
+		return configMap.Set("log_level=4") // Warn level and above
+	case log.INFO:
+		return configMap.Set("log_level=6") // Info level and above
 	case log.DEBUG:
 		return configMap.Set("debug=broker, topic, msg, protocol")
 	case log.TRACE, log.UNSAFE_TRACE:
 		return configMap.Set("debug=all")
 	}
-	return nil
+	return configMap.Set("log_level=3") // error level and above
 }
 
 func SetConsumerDebugOption(configMap *ckgo.ConfigMap) error {
+	// Note: log_levels are based on syslog levels
 	switch log.CliLogger.Level {
+	case log.WARN:
+		return configMap.Set("log_level=4") // Warn level and above
+	case log.INFO:
+		return configMap.Set("log_level=6") // Info level and above
 	case log.DEBUG:
 		return configMap.Set("debug=broker, topic, msg, protocol, consumer, cgrp, fetch")
 	case log.TRACE, log.UNSAFE_TRACE:
 		return configMap.Set("debug=all")
 	}
-	return nil
+	return configMap.Set("log_level=3") // error level and above
 }
 
 func newProducerWithOverwrittenConfigs(configMap *ckgo.ConfigMap, configPath string, configStrings []string) (*ckgo.Producer, error) {