Merge pull request #172 from nanic/feat/json-produce-support

Feat/json produce support

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- [#171](https://github.com/deviceinsight/kafkactl/issues/171) Support for reset command to reset offset to offset from provided datetime
+- [#172](https://github.com/deviceinsight/kafkactl/issues/172) Support for json format in produce command
+
 ## 3.4.0 - 2023-09-25
 
 ### Fixed

README.md

@@ -432,6 +432,14 @@ kafkactl produce my-topic --separator=# --lineSeparator=|| --file=myfile
 **NOTE:** if the file was generated with `kafkactl consume --print-keys --print-timestamps my-topic` the produce
 command is able to detect the message timestamp in the input and will ignore it.
 
+It is also possible to produce messages in json format:
+
+```bash
+# each line in myfile.json is expected to contain a json object with fields key, value
+kafkactl produce my-topic --file=myfile.json --input-format=json
+cat myfile.json | kafkactl produce my-topic --input-format=json
+```
+
 the number of messages produced per second can be controlled with the `--rate` parameter:
 
 ```bash

cmd/produce/produce.go

@@ -36,6 +36,7 @@ func NewProduceCmd() *cobra.Command {
 	cmdProduce.Flags().StringVarP(&flags.Value, "value", "v", "", "value to produce")
 	cmdProduce.Flags().BoolVarP(&flags.NullValue, "null-value", "", false, "produce a null value (can be used instead of providing a value with --value)")
 	cmdProduce.Flags().StringVarP(&flags.File, "file", "f", "", "file to read input from")
+	cmdProduce.Flags().StringVarP(&flags.InputFormat, "input-format", "", "", "input format. One of: csv,json (default is csv)")
 	cmdProduce.Flags().StringArrayVarP(&flags.Headers, "header", "H", flags.Headers, "headers in format `key:value`")
 	cmdProduce.Flags().StringVarP(&flags.Separator, "separator", "S", "", "separator to split key and value from stdin or file")
 	cmdProduce.Flags().StringVarP(&flags.LineSeparator, "lineSeparator", "L", "\n", "separator to split multiple messages from stdin or file")

cmd/produce/produce_test.go

@@ -367,6 +367,95 @@ func TestProduceProtoFileIntegration(t *testing.T) {
 	testutil.AssertEquals(t, value, string(actualValue))
 }
 
+func TestProduceWithCSVFileIntegration(t *testing.T) {
+
+	testutil.StartIntegrationTest(t)
+	topic := testutil.CreateTopic(t, "produce-topic-csv")
+	kafkaCtl := testutil.CreateKafkaCtlCommand()
+
+	dataFilePath := filepath.Join(testutil.RootDir, "testutil", "testdata")
+
+	if _, err := kafkaCtl.Execute("produce", topic, "--separator", ",",
+		"--file", filepath.Join(dataFilePath, "msg.csv")); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, "3 messages produced", kafkaCtl.GetStdOut())
+
+	if _, err := kafkaCtl.Execute("consume", topic, "--from-beginning", "--print-keys", "--exit"); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, "1#a\n2#b\n3#c", kafkaCtl.GetStdOut())
+}
+
+func TestProduceWithCSVFileWithTimestampsFirstColumnIntegration(t *testing.T) {
+
+	testutil.StartIntegrationTest(t)
+	topic := testutil.CreateTopic(t, "produce-topic-csv")
+	kafkaCtl := testutil.CreateKafkaCtlCommand()
+
+	dataFilePath := filepath.Join(testutil.RootDir, "testutil", "testdata")
+
+	if _, err := kafkaCtl.Execute("produce", topic, "--separator", ",",
+		"--file", filepath.Join(dataFilePath, "msg-ts1.csv")); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, "3 messages produced", kafkaCtl.GetStdOut())
+
+	if _, err := kafkaCtl.Execute("consume", topic, "--from-beginning", "--print-keys", "--exit"); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, "1#a\n2#b\n3#c", kafkaCtl.GetStdOut())
+}
+
+func TestProduceWithCSVFileWithTimestampsSecondColumnIntegration(t *testing.T) {
+
+	testutil.StartIntegrationTest(t)
+	topic := testutil.CreateTopic(t, "produce-topic-csv")
+	kafkaCtl := testutil.CreateKafkaCtlCommand()
+
+	dataFilePath := filepath.Join(testutil.RootDir, "testutil", "testdata")
+
+	if _, err := kafkaCtl.Execute("produce", topic, "--separator", ",",
+		"--file", filepath.Join(dataFilePath, "msg-ts2.csv")); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, "3 messages produced", kafkaCtl.GetStdOut())
+
+	if _, err := kafkaCtl.Execute("consume", topic, "--from-beginning", "--print-keys", "--exit"); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, "1#a\n2#b\n3#c", kafkaCtl.GetStdOut())
+}
+
+func TestProduceWithJSONFileIntegration(t *testing.T) {
+
+	testutil.StartIntegrationTest(t)
+	topic := testutil.CreateTopic(t, "produce-topic-json")
+	kafkaCtl := testutil.CreateKafkaCtlCommand()
+
+	dataFilePath := filepath.Join(testutil.RootDir, "testutil", "testdata")
+
+	if _, err := kafkaCtl.Execute("produce", topic,
+		"--file", filepath.Join(dataFilePath, "msg.json"),
+		"--input-format", "json"); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, "3 messages produced", kafkaCtl.GetStdOut())
+
+	if _, err := kafkaCtl.Execute("consume", topic, "--from-beginning", "--print-keys", "--exit"); err != nil {
+		t.Fatalf("failed to execute command: %v", err)
+	}
+
+	testutil.AssertEquals(t, "1#a\n2#b\n3#c", kafkaCtl.GetStdOut())
+}
+
 func TestProduceProtoFileWithOnlyKeyEncodedIntegration(t *testing.T) {
 	testutil.StartIntegrationTest(t)
 

internal/producer/input/csv_parser.go

@@ -0,0 +1,72 @@
+package input
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/deviceinsight/kafkactl/output"
+	"github.com/deviceinsight/kafkactl/util"
+	"github.com/pkg/errors"
+)
+
+const defaultKeyColumnIdx = 0
+const defaultValueColumnIdx = 1
+const defaultColumnCount = 2
+
+type csvParser struct {
+	key            string
+	separator      string
+	keyColumnIdx   int
+	valueColumnIdx int
+	columnCount    int
+}
+
+func NewCsvParser(key string, separator string) Parser {
+	return &csvParser{
+		key:            key,
+		separator:      separator,
+		keyColumnIdx:   defaultKeyColumnIdx,
+		valueColumnIdx: defaultValueColumnIdx,
+		columnCount:    defaultColumnCount,
+	}
+}
+
+func (p *csvParser) ParseLine(line string) (Message, error) {
+
+	if p.separator == "" {
+		return Message{Key: p.key, Value: line}, nil
+	}
+
+	input := strings.Split(line, util.ConvertControlChars(p.separator))
+	if len(input) < 2 {
+		return Message{}, fmt.Errorf("the provided input does not contain the separator %s", p.separator)
+	} else if len(input) == 3 && p.columnCount == defaultColumnCount {
+		// lazy resolving of column indices
+		var err error
+		p.keyColumnIdx, p.valueColumnIdx, p.columnCount, err = resolveColumns(input)
+		if err != nil {
+			return Message{}, err
+		}
+	} else if len(input) != p.columnCount {
+		return Message{}, fmt.Errorf("line contains unexpected amount of separators:\n%s", line)
+	}
+
+	return Message{input[p.keyColumnIdx], input[p.valueColumnIdx]}, nil
+}
+
+func resolveColumns(line []string) (keyColumnIdx, valueColumnIdx, columnCount int, err error) {
+	if isTimestamp(line[0]) {
+		output.Warnf("assuming column 0 to be message timestamp. Column will be ignored")
+		return 1, 2, 3, nil
+	} else if isTimestamp(line[1]) {
+		output.Warnf("assuming column 1 to be message timestamp. Column will be ignored")
+		return 0, 2, 3, nil
+	}
+	return -1, -1, -1, errors.Errorf("line contains unexpected amount of separators:\n%s", line)
+}
+
+func isTimestamp(value string) bool {
+	_, e := time.Parse(time.RFC3339, value)
+	return e == nil
+}

internal/producer/input/json_parser.go

@@ -0,0 +1,24 @@
+package input
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+type jsonParser struct {
+}
+
+func NewJSONParser() Parser {
+	return &jsonParser{}
+}
+
+func (p *jsonParser) ParseLine(line string) (Message, error) {
+
+	var message Message
+
+	if err := json.Unmarshal([]byte(line), &message); err != nil {
+		return message, fmt.Errorf("can't unmarshal line: %w", err)
+	}
+
+	return message, nil
+}

internal/producer/input/parser.go

@@ -0,0 +1,10 @@
+package input
+
+type Message struct {
+	Key   string `json:"key"`
+	Value string `json:"value"`
+}
+
+type Parser interface {
+	ParseLine(line string) (Message, error)
+}

internal/producer/producer-operation.go

@@ -7,11 +7,12 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
-	"time"
 
 	"github.com/IBM/sarama"
 	"github.com/deviceinsight/kafkactl/internal"
+	"github.com/deviceinsight/kafkactl/internal/producer/input"
 	"github.com/deviceinsight/kafkactl/output"
+	"github.com/deviceinsight/kafkactl/util"
 	"github.com/pkg/errors"
 	"go.uber.org/ratelimit"
 )
@@ -24,6 +25,7 @@ type Flags struct {
 	Separator          string
 	LineSeparator      string
 	File               string
+	InputFormat        string
 	Key                string
 	Value              string
 	NullValue          bool
@@ -113,13 +115,10 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 		}
 	}()
 
-	var key string
-	var value string
+	var inputMessage input.Message
 
 	if flags.Key != "" && flags.Separator != "" {
 		return errors.New("parameters --key and --separator cannot be used together")
-	} else if flags.Key != "" {
-		key = flags.Key
 	}
 
 	if flags.NullValue && flags.Value != "" {
@@ -131,9 +130,9 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 		var message *sarama.ProducerMessage
 
 		if flags.NullValue {
-			message, err = serializers.Serialize([]byte(key), nil, flags)
+			message, err = serializers.Serialize([]byte(flags.Key), nil, flags)
 		} else {
-			message, err = serializers.Serialize([]byte(key), []byte(flags.Value), flags)
+			message, err = serializers.Serialize([]byte(flags.Key), []byte(flags.Value), flags)
 		}
 
 		if err != nil {
@@ -157,9 +156,7 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 		}()
 
 		messageCount := 0
-		keyColumnIdx := 0
-		valueColumnIdx := 1
-		columnCount := 2
+
 		// print an empty line that will be replaced when updating the status
 		output.Statusf("")
 
@@ -173,6 +170,7 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 		}
 
 		var inputReader io.Reader
+		var inputParser input.Parser
 
 		if flags.File != "" {
 			inputReader, err = os.Open(flags.File)
@@ -183,11 +181,17 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 			inputReader = os.Stdin
 		}
 
+		if flags.InputFormat == "json" {
+			inputParser = input.NewJSONParser()
+		} else {
+			inputParser = input.NewCsvParser(flags.Key, flags.Separator)
+		}
+
 		scanner := bufio.NewScanner(inputReader)
 		scanner.Buffer(make([]byte, 0, config.Producer.MaxMessageBytes), config.Producer.MaxMessageBytes)
 
 		if len(flags.LineSeparator) > 0 && flags.LineSeparator != "\n" {
-			scanner.Split(splitAt(convertControlChars(flags.LineSeparator)))
+			scanner.Split(splitAt(util.ConvertControlChars(flags.LineSeparator)))
 		}
 
 		for scanner.Scan() {
@@ -207,26 +211,12 @@ func (operation *Operation) Produce(topic string, flags Flags) error {
 				continue
 			}
 
-			if flags.Separator != "" {
-				input := strings.Split(line, convertControlChars(flags.Separator))
-				if len(input) < 2 {
-					return failWithMessageCount(messageCount, "the provided input does not contain the separator %s", flags.Separator)
-				} else if len(input) == 3 && messageCount == 0 {
-					keyColumnIdx, valueColumnIdx, columnCount, err = resolveColumns(input)
-					if err != nil {
-						return failWithMessageCount(messageCount, err.Error())
-					}
-				} else if len(input) != columnCount {
-					return failWithMessageCount(messageCount, "line contains unexpected amount of separators:\n%s", line)
-				}
-				key = input[keyColumnIdx]
-				value = input[valueColumnIdx]
-			} else {
-				value = line
+			if inputMessage, err = inputParser.ParseLine(line); err != nil {
+				return failWithMessageCount(messageCount, err.Error())
 			}
 
 			messageCount++
-			message, err := serializers.Serialize([]byte(key), []byte(value), flags)
+			message, err := serializers.Serialize([]byte(inputMessage.Key), []byte(inputMessage.Value), flags)
 			if err != nil {
 				return errors.Wrap(err, "Failed to produce message")
 			}
@@ -325,29 +315,6 @@ func parsePartitioner(partitioner string, flags Flags) (sarama.PartitionerConstr
 	}
 }
 
-func convertControlChars(value string) string {
-	value = strings.Replace(value, "\\n", "\n", -1)
-	value = strings.Replace(value, "\\r", "\r", -1)
-	value = strings.Replace(value, "\\t", "\t", -1)
-	return value
-}
-
-func resolveColumns(line []string) (keyColumnIdx, valueColumnIdx, columnCount int, err error) {
-	if isTimestamp(line[0]) {
-		output.Warnf("assuming column 0 to be message timestamp. Column will be ignored")
-		return 1, 2, 3, nil
-	} else if isTimestamp(line[1]) {
-		output.Warnf("assuming column 1 to be message timestamp. Column will be ignored")
-		return 0, 2, 3, nil
-	}
-	return -1, -1, -1, errors.Errorf("line contains unexpected amount of separators:\n%s", line)
-}
-
-func isTimestamp(value string) bool {
-	_, e := time.Parse(time.RFC3339, value)
-	return e == nil
-}
-
 func failWithMessageCount(messageCount int, errorMessage string, args ...interface{}) error {
 	output.Infof("\r%d messages produced", messageCount)
 	return errors.Errorf(errorMessage, args...)

testutil/testdata/msg-ts1.csv

@@ -0,0 +1,3 @@
+2006-01-02T15:04:05.000Z,1,a
+2006-01-02T16:04:05.000Z,2,b
+2006-01-02T17:04:05.000Z,3,c

testutil/testdata/msg-ts2.csv

@@ -0,0 +1,3 @@
+1,2006-01-02T15:04:05.000Z,a
+2,2006-01-02T16:04:05.000Z,b
+3,2006-01-02T17:04:05.000Z,c