feat: write json and string methods for producer and consumer configs…

balancer.go

@@ -27,3 +27,22 @@ func GetBalancerReferenceHash() Balancer {
 func GetBalancerRoundRobin() Balancer {
 	return &kafka.RoundRobin{}
 }
+
+func GetBalancerString(balancer Balancer) string {
+	switch balancer.(type) {
+	case *kafka.CRC32Balancer:
+		return "CRC32Balancer"
+	case *kafka.Hash:
+		return "Hash"
+	case *kafka.LeastBytes:
+		return "LeastBytes"
+	case *kafka.Murmur2Balancer:
+		return "Murmur2Balancer"
+	case *kafka.ReferenceHash:
+		return "ReferenceHash"
+	case *kafka.RoundRobin:
+		return "RoundRobin"
+	default:
+		return "Unknown"
+	}
+}

balancer_test.go

@@ -64,3 +64,54 @@ func TestGetBalancerRoundRobinh(t *testing.T) {
 		t.Errorf("Expected *kafka.RoundRobin, got %s", reflect.TypeOf(balancer).String())
 	}
 }
+
+func TestGetBalancerString(t *testing.T) {
+	tests := []struct {
+		name     string
+		balancer Balancer
+		want     string
+	}{
+		{
+			name:     "Should_Return_CRC32Balancer",
+			balancer: GetBalancerCRC32(),
+			want:     "CRC32Balancer",
+		},
+		{
+			name:     "Should_Return_Hash",
+			balancer: GetBalancerHash(),
+			want:     "Hash",
+		},
+		{
+			name:     "Should_Return_LeastBytes",
+			balancer: GetBalancerLeastBytes(),
+			want:     "LeastBytes",
+		},
+		{
+			name:     "Should_Return_Murmur2Balancer",
+			balancer: GetBalancerMurmur2Balancer(),
+			want:     "Murmur2Balancer",
+		},
+		{
+			name:     "Should_Return_ReferenceHash",
+			balancer: GetBalancerReferenceHash(),
+			want:     "ReferenceHash",
+		},
+		{
+			name:     "Should_Return_RoundRobin",
+			balancer: GetBalancerRoundRobin(),
+			want:     "RoundRobin",
+		},
+		{
+			name:     "Should_Return_Unknown",
+			balancer: nil,
+			want:     "Unknown",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := GetBalancerString(tt.balancer); got != tt.want {
+				t.Errorf("GetBalancerString() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

consumer_config.go

@@ -1,6 +1,11 @@
 package kafka
 
 import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"regexp"
+	"strings"
 	"time"
 
 	"github.com/segmentio/kafka-go"
@@ -61,6 +66,51 @@ type ConsumerConfig struct {
 	MetricPrefix string
 }
 
+func (cfg RetryConfiguration) JSON() string {
+	return fmt.Sprintf(`{"Brokers": ["%s"], "Topic": %q, "StartTimeCron": %q, "WorkDuration": %q, `+
+		`"MaxRetry": %d, "VerifyTopicOnStartup": %t, "Rack": %q}`,
+		strings.Join(cfg.Brokers, "\", \""), cfg.Topic, cfg.StartTimeCron,
+		cfg.WorkDuration, cfg.MaxRetry, cfg.VerifyTopicOnStartup, cfg.Rack)
+}
+
+func (cfg *BatchConfiguration) JSON() string {
+	if cfg == nil {
+		return "{}"
+	}
+	return fmt.Sprintf(`{"MessageGroupLimit": %d}`, cfg.MessageGroupLimit)
+}
+
+func (cfg ReaderConfig) JSON() string {
+	return fmt.Sprintf(`{"Brokers": ["%s"], "GroupId": %q, "GroupTopics": ["%s"], `+
+		`"MaxWait": %q, "CommitInterval": %q, "StartOffset": %q}`,
+		strings.Join(cfg.Brokers, "\", \""), cfg.GroupID, strings.Join(cfg.GroupTopics, "\", \""),
+		cfg.MaxWait, cfg.CommitInterval, kcronsumer.ToStringOffset(cfg.StartOffset))
+}
+
+func (cfg *ConsumerConfig) JSON() string {
+	if cfg == nil {
+		return "{}"
+	}
+	return fmt.Sprintf(`{"ClientID": %q, "Reader": %s, "BatchConfiguration": %s, "MessageGroupDuration": %q, `+
+		`"TransactionalRetry": %t, "Concurrency": %d, "RetryEnabled": %t, "RetryConfiguration": %s, `+
+		`"VerifyTopicOnStartup": %t, "Rack": %q, "SASL": %s, "TLS": %s}`,
+		cfg.ClientID, cfg.Reader.JSON(), cfg.BatchConfiguration.JSON(),
+		cfg.MessageGroupDuration, *cfg.TransactionalRetry, cfg.Concurrency,
+		cfg.RetryEnabled, cfg.RetryConfiguration.JSON(), cfg.VerifyTopicOnStartup,
+		cfg.Rack, cfg.SASL.JSON(), cfg.TLS.JSON())
+}
+
+func (cfg *ConsumerConfig) JSONPretty() string {
+	return jsonPretty(cfg.JSON())
+}
+
+func (cfg *ConsumerConfig) String() string {
+	re := regexp.MustCompile(`"(\w+)"\s*:`)
+	modifiedString := re.ReplaceAllString(cfg.JSON(), `$1:`)
+	modifiedString = modifiedString[1 : len(modifiedString)-1]
+	return modifiedString
+}
+
 func (cfg *ConsumerConfig) newCronsumerConfig() *kcronsumer.Config {
 	cronsumerCfg := kcronsumer.Config{
 		MetricPrefix: cfg.RetryConfiguration.MetricPrefix,
@@ -266,3 +316,12 @@ func (cfg *ConsumerConfig) setDefaults() {
 func NewBoolPtr(value bool) *bool {
 	return &value
 }
+
+func jsonPretty(jsonString string) string {
+	var out bytes.Buffer
+	err := json.Indent(&out, []byte(jsonString), "", "\t")
+	if err != nil {
+		return jsonString
+	}
+	return out.String()
+}

consumer_config_test.go

@@ -189,3 +189,219 @@ func TestConsumerConfig_getTopics(t *testing.T) {
 		}
 	})
 }
+
+func Test_jsonPretty(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		expected string
+	}{
+		{
+			name:     "Simple JSON",
+			input:    `{"key1":"value1","key2":2}`,
+			expected: "{\n\t\"key1\": \"value1\",\n\t\"key2\": 2\n}",
+		},
+		{
+			name:  "Nested JSON",
+			input: `{"key1":"value1","key2":{"nestedKey1":1,"nestedKey2":2},"key3":[1,2,3]}`,
+			expected: "{\n\t\"key1\": \"value1\",\n\t\"" +
+				"key2\": {\n\t\t\"nestedKey1\": 1,\n\t\t\"nestedKey2\": 2\n\t},\n\t\"" +
+				"key3\": [\n\t\t1,\n\t\t2,\n\t\t3\n\t]\n}",
+		},
+		{
+			name:     "Invalid JSON",
+			input:    `{"key1": "value1", "key2": 2`,
+			expected: `{"key1": "value1", "key2": 2`,
+		},
+		{
+			name:     "Empty JSON",
+			input:    ``,
+			expected: ``,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := jsonPretty(tt.input)
+			if got != tt.expected {
+				t.Errorf("jsonPretty() = %v, want %v", got, tt.expected)
+			}
+		})
+	}
+}
+
+func TestConsumerConfig_JSON(t *testing.T) {
+	t.Run("Should_Convert_Nil_Config_To_Json", func(t *testing.T) {
+		// Given
+		var config *ConsumerConfig
+		expected := "{}"
+		// When
+		result := config.JSON()
+		// Then
+		if result != expected {
+			t.Fatal("result must be equal to expected")
+		}
+	})
+	t.Run("Should_Convert_To_Json", func(t *testing.T) {
+		// Given
+		expected := "{\"ClientID\": \"test-consumer-client-id\", \"Reader\": {\"Brokers\": [\"broker-1.test.com\", \"broker-2.test.com\"], " +
+			"\"GroupId\": \"test-consumer.0\", \"GroupTopics\": [\"test-updated.0\"], \"MaxWait\": \"2s\", " +
+			"\"CommitInterval\": \"1s\", \"StartOffset\": \"earliest\"}, \"BatchConfiguration\": {\"MessageGroupLimit\": 100}, " +
+			"\"MessageGroupDuration\": \"20ns\", \"TransactionalRetry\": false, \"Concurrency\": 10, \"RetryEnabled\": true, " +
+			"\"RetryConfiguration\": {\"Brokers\": [\"broker-1.test.com\", \"broker-2.test.com\"], \"Topic\": \"test-exception.0\", " +
+			"\"StartTimeCron\": \"*/2 * * * *\", \"WorkDuration\": \"1m0s\", \"MaxRetry\": 3, \"VerifyTopicOnStartup\": true, \"Rack\": \"\"}, " +
+			"\"VerifyTopicOnStartup\": true, \"Rack\": \"stage\", " +
+			"\"SASL\": {\"Mechanism\": \"scram\", \"Username\": \"user\", \"Password\": \"pass\"}, " +
+			"\"TLS\": {\"RootCAPath\": \"resources/ca\", \"IntermediateCAPath\": \"resources/intCa\"}}"
+		// When
+		result := getConsumerConfigExample().JSON()
+		// Then
+		if result != expected {
+			t.Fatal("result must be equal to expected")
+		}
+	})
+	t.Run("Should_Convert_To_Json_Without_Inner_Object", func(t *testing.T) {
+		// Given
+		expected := "{\"ClientID\": \"test-consumer-client-id\", \"Reader\": {\"Brokers\": [\"\"], \"GroupId\": \"\", " +
+			"\"GroupTopics\": [\"\"], \"MaxWait\": \"0s\", \"CommitInterval\": \"0s\", \"StartOffset\": \"earliest\"}, " +
+			"\"BatchConfiguration\": {}, \"MessageGroupDuration\": \"20ns\", \"TransactionalRetry\": false, \"Concurrency\": 10, " +
+			"\"RetryEnabled\": true, \"RetryConfiguration\": {\"Brokers\": [\"\"], \"Topic\": \"\", \"StartTimeCron\": \"\", " +
+			"\"WorkDuration\": \"0s\", \"MaxRetry\": 0, \"VerifyTopicOnStartup\": false, \"Rack\": \"\"}, \"VerifyTopicOnStartup\": true, " +
+			"\"Rack\": \"stage\", \"SASL\": {}, \"TLS\": {}}"
+		// When
+		result := getConsumerConfigWithoutInnerObjectExample().JSON()
+		// Then
+		if result != expected {
+			t.Fatal("result must be equal to expected")
+		}
+	})
+}
+
+func TestConsumerConfig_String(t *testing.T) {
+	t.Run("Should_Convert_To_String", func(t *testing.T) {
+		// Given
+		expected := "ClientID: \"test-consumer-client-id\", Reader: {Brokers: [\"broker-1.test.com\", \"broker-2.test.com\"], " +
+			"GroupId: \"test-consumer.0\", GroupTopics: [\"test-updated.0\"], MaxWait: \"2s\", CommitInterval: \"1s\", " +
+			"StartOffset: \"earliest\"}, BatchConfiguration: {MessageGroupLimit: 100}, MessageGroupDuration: \"20ns\", " +
+			"TransactionalRetry: false, Concurrency: 10, RetryEnabled: true, " +
+			"RetryConfiguration: {Brokers: [\"broker-1.test.com\", \"broker-2.test.com\"], Topic: \"test-exception.0\", " +
+			"StartTimeCron: \"*/2 * * * *\", WorkDuration: \"1m0s\", MaxRetry: 3, VerifyTopicOnStartup: true, Rack: \"\"}, " +
+			"VerifyTopicOnStartup: true, Rack: \"stage\", SASL: {Mechanism: \"scram\", Username: \"user\", Password: \"pass\"}, " +
+			"TLS: {RootCAPath: \"resources/ca\", IntermediateCAPath: \"resources/intCa\"}"
+		// When
+		result := getConsumerConfigExample().String()
+		// Then
+		if result != expected {
+			t.Fatal("result must be equal to expected")
+		}
+	})
+	t.Run("Should_Convert_To_String_Without_Inner_Object", func(t *testing.T) {
+		// Given
+		expected := "ClientID: \"test-consumer-client-id\", Reader: {Brokers: [\"\"], GroupId: \"\", " +
+			"GroupTopics: [\"\"], MaxWait: \"0s\", CommitInterval: \"0s\", StartOffset: \"earliest\"}, " +
+			"BatchConfiguration: {}, MessageGroupDuration: \"20ns\", TransactionalRetry: false, Concurrency: 10, " +
+			"RetryEnabled: true, RetryConfiguration: {Brokers: [\"\"], Topic: \"\", StartTimeCron: \"\", WorkDuration: \"0s\", " +
+			"MaxRetry: 0, VerifyTopicOnStartup: false, Rack: \"\"}, VerifyTopicOnStartup: true, Rack: \"stage\", SASL: {}, TLS: {}"
+		// When
+		result := getConsumerConfigWithoutInnerObjectExample().String()
+		// Then
+		if result != expected {
+			t.Fatal("result must be equal to expected")
+		}
+	})
+}
+
+func TestConsumerConfig_JSONPretty(t *testing.T) {
+	t.Run("Should_Convert_To_Pretty_Json", func(t *testing.T) {
+		// Given
+		expected := "{\n\t\"ClientID\": \"test-consumer-client-id\",\n\t\"Reader\": {\n\t\t\"" +
+			"Brokers\": [\n\t\t\t\"broker-1.test.com\",\n\t\t\t\"broker-2.test.com\"\n\t\t],\n\t\t\"" +
+			"GroupId\": \"test-consumer.0\",\n\t\t\"GroupTopics\": [\n\t\t\t\"test-updated.0\"\n\t\t],\n\t\t\"" +
+			"MaxWait\": \"2s\",\n\t\t\"CommitInterval\": \"1s\",\n\t\t\"StartOffset\": \"earliest\"\n\t},\n\t\"" +
+			"BatchConfiguration\": {\n\t\t\"MessageGroupLimit\": 100\n\t},\n\t\"MessageGroupDuration\": \"20ns\",\n\t\"" +
+			"TransactionalRetry\": false,\n\t\"Concurrency\": 10,\n\t\"RetryEnabled\": true,\n\t\"" +
+			"RetryConfiguration\": {\n\t\t\"Brokers\": [\n\t\t\t\"broker-1.test.com\",\n\t\t\t\"broker-2.test.com\"\n\t\t],\n\t\t\"" +
+			"Topic\": \"test-exception.0\",\n\t\t\"StartTimeCron\": \"*/2 * * * *\",\n\t\t\"WorkDuration\": \"1m0s\",\n\t\t\"" +
+			"MaxRetry\": 3,\n\t\t\"VerifyTopicOnStartup\": true,\n\t\t\"Rack\": \"\"\n\t},\n\t\"" +
+			"VerifyTopicOnStartup\": true,\n\t\"Rack\": \"stage\",\n\t\"" +
+			"SASL\": {\n\t\t\"Mechanism\": \"scram\",\n\t\t\"Username\": \"user\",\n\t\t\"Password\": \"pass\"\n\t},\n\t\"" +
+			"TLS\": {\n\t\t\"RootCAPath\": \"resources/ca\",\n\t\t\"IntermediateCAPath\": \"resources/intCa\"\n\t}\n}"
+		// When
+		result := getConsumerConfigExample().JSONPretty()
+		// Then
+		if result != expected {
+			t.Fatal("result must be equal to expected")
+		}
+	})
+	t.Run("Should_Convert_To_Pretty_Json_Without_Inner_Object", func(t *testing.T) {
+		// Given
+		expected := "{\n\t\"ClientID\": \"test-consumer-client-id\",\n\t\"" +
+			"Reader\": {\n\t\t\"Brokers\": [\n\t\t\t\"\"\n\t\t],\n\t\t\"GroupId\": \"\",\n\t\t\"" +
+			"GroupTopics\": [\n\t\t\t\"\"\n\t\t],\n\t\t\"MaxWait\": \"0s\",\n\t\t\"CommitInterval\": \"0s\",\n\t\t\"" +
+			"StartOffset\": \"earliest\"\n\t},\n\t\"BatchConfiguration\": {},\n\t\"" +
+			"MessageGroupDuration\": \"20ns\",\n\t\"TransactionalRetry\": false,\n\t\"Concurrency\": 10,\n\t\"" +
+			"RetryEnabled\": true,\n\t\"RetryConfiguration\": {\n\t\t\"Brokers\": [\n\t\t\t\"\"\n\t\t],\n\t\t\"" +
+			"Topic\": \"\",\n\t\t\"StartTimeCron\": \"\",\n\t\t\"WorkDuration\": \"0s\",\n\t\t\"MaxRetry\": 0,\n\t\t\"" +
+			"VerifyTopicOnStartup\": false,\n\t\t\"Rack\": \"\"\n\t},\n\t\"VerifyTopicOnStartup\": true,\n\t\"" +
+			"Rack\": \"stage\",\n\t\"SASL\": {},\n\t\"TLS\": {}\n}"
+		// When
+		result := getConsumerConfigWithoutInnerObjectExample().JSONPretty()
+		// Then
+		if result != expected {
+			t.Fatal("result must be equal to expected")
+		}
+	})
+}
+
+func getConsumerConfigExample() *ConsumerConfig {
+	return &ConsumerConfig{
+		Rack:     "stage",
+		ClientID: "test-consumer-client-id",
+		Reader: ReaderConfig{
+			Brokers:        []string{"broker-1.test.com", "broker-2.test.com"},
+			GroupID:        "test-consumer.0",
+			GroupTopics:    []string{"test-updated.0"},
+			MaxWait:        2 * time.Second,
+			CommitInterval: time.Second,
+		},
+		BatchConfiguration: &BatchConfiguration{
+			MessageGroupLimit: 100,
+		},
+		MessageGroupDuration: 20,
+		TransactionalRetry:   NewBoolPtr(false),
+		Concurrency:          10,
+		RetryEnabled:         true,
+		RetryConfiguration: RetryConfiguration{
+			Brokers:              []string{"broker-1.test.com", "broker-2.test.com"},
+			Topic:                "test-exception.0",
+			StartTimeCron:        "*/2 * * * *",
+			WorkDuration:         time.Minute * 1,
+			MaxRetry:             3,
+			VerifyTopicOnStartup: true,
+		},
+		VerifyTopicOnStartup: true,
+		TLS: &TLSConfig{
+			RootCAPath:         "resources/ca",
+			IntermediateCAPath: "resources/intCa",
+		},
+		SASL: &SASLConfig{
+			Type:     "scram",
+			Username: "user",
+			Password: "pass",
+		},
+	}
+}
+
+func getConsumerConfigWithoutInnerObjectExample() *ConsumerConfig {
+	return &ConsumerConfig{
+		Rack:                 "stage",
+		ClientID:             "test-consumer-client-id",
+		Reader:               ReaderConfig{},
+		MessageGroupDuration: 20,
+		TransactionalRetry:   NewBoolPtr(false),
+		Concurrency:          10,
+		RetryEnabled:         true,
+		RetryConfiguration:   RetryConfiguration{},
+		VerifyTopicOnStartup: true,
+	}
+}

mechanism.go

@@ -1,6 +1,8 @@
 package kafka
 
 import (
+	"fmt"
+
 	"github.com/segmentio/kafka-go/sasl"
 	"github.com/segmentio/kafka-go/sasl/plain"
 	"github.com/segmentio/kafka-go/sasl/scram"
@@ -37,3 +39,10 @@ func (s *SASLConfig) plain() sasl.Mechanism {
 func (s *SASLConfig) IsEmpty() bool {
 	return s == nil
 }
+
+func (s *SASLConfig) JSON() string {
+	if s == nil {
+		return "{}"
+	}
+	return fmt.Sprintf(`{"Mechanism": %q, "Username": %q, "Password": %q}`, s.Type, s.Username, s.Password)
+}