Fix --headers flag bug in local kafka topic produce (#2819)

confluentinc · Jun 18, 2024 · def8f09 · def8f09
1 parent ea012ec
commit def8f09
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 15 deletions.
diff --git a/internal/kafka/command_topic.go b/internal/kafka/command_topic.go
@@ -237,7 +237,7 @@ func ProduceToTopic(cmd *cobra.Command, keyMetaInfo []byte, valueMetaInfo []byte
  break
  }
 
- message, err := GetProduceMessage(cmd, keyMetaInfo, valueMetaInfo, topic, data, keySerializer, valueSerializer)
+ message, err := getProduceMessage(cmd, keyMetaInfo, valueMetaInfo, topic, data, keySerializer, valueSerializer)
  if err != nil {
  return err
  }

diff --git a/internal/kafka/command_topic_produce.go b/internal/kafka/command_topic_produce.go
@@ -408,7 +408,7 @@ func PrepareInputChannel(scanErr *error) (chan string, func()) {
  }
 }
 
-func GetProduceMessage(cmd *cobra.Command, keyMetaInfo, valueMetaInfo []byte, topic, data string, keySerializer, valueSerializer serdes.SerializationProvider) (*ckafka.Message, error) {
+func getProduceMessage(cmd *cobra.Command, keyMetaInfo, valueMetaInfo []byte, topic, data string, keySerializer, valueSerializer serdes.SerializationProvider) (*ckafka.Message, error) {
  parseKey, err := cmd.Flags().GetBool("parse-key")
  if err != nil {
  return nil, err
@@ -419,16 +419,6 @@ func GetProduceMessage(cmd *cobra.Command, keyMetaInfo, valueMetaInfo []byte, to
  return nil, err
  }
 
- headers, err := cmd.Flags().GetStringSlice("headers")
- if err != nil {
- return nil, err
- }
-
- parsedHeaders, err := parseHeaders(headers, delimiter)
- if err != nil {
- return nil, err
- }
-
  key, value, err := serializeMessage(keyMetaInfo, valueMetaInfo, data, delimiter, parseKey, keySerializer, valueSerializer)
  if err != nil {
  return nil, err
@@ -439,9 +429,18 @@ func GetProduceMessage(cmd *cobra.Command, keyMetaInfo, valueMetaInfo []byte, to
  Topic: &topic,
  Partition: ckafka.PartitionAny,
  },
- Key: key,
- Value: value,
- Headers: parsedHeaders,
+ Key: key,
+ Value: value,
+ }
+
+ // This error is intentionally ignored because `confluent local kafka topic produce` does not define this flag
+ headers, _ := cmd.Flags().GetStringSlice("headers")
+ if headers != nil {
+ parsedHeaders, err := parseHeaders(headers, delimiter)
+ if err != nil {
+ return nil, err
+ }
+ message.Headers = parsedHeaders
  }
 
  return message, nil