fix(kafka topic): update regexp for topic name

pkg/kafka/topic/validators.go

@@ -13,7 +13,7 @@ import (
 )
 
 const (
-	legalNameChars = "^[a-zA-Z0-9\\_\\-]+$"
+	legalNameChars = `[a-zA-Z0-9._-]`
 	maxNameLength  = 249
 	minPartitions  = 1
 	maxPartitions  = 100
@@ -32,13 +32,17 @@ func ValidateName(val interface{}) error {
 		return fmt.Errorf("topic name cannot exceed %v characters", maxNameLength)
 	}
 
+	if (name == ".") || (name == "..") {
+		return fmt.Errorf(`invalid topic name "%v"; can't be "." or ".."`, name)
+	}
+
 	matched, _ := regexp.Match(legalNameChars, []byte(name))
 
 	if matched {
 		return nil
 	}
 
-	return fmt.Errorf(`invalid topic name "%v"; only letters (Aa-Zz), numbers, "_" and "-" are accepted`, name)
+	return fmt.Errorf(`invalid topic name "%v"; only letters (Aa-Zz), numbers, "_", "." and "-" are accepted`, name)
 }
 
 // ValidatePartitionsN performs validation on the number of partitions v

pkg/kafka/topic/validators_test.go

@@ -0,0 +1,94 @@
+package topic
+
+import "testing"
+
+func TestValidateName(t *testing.T) {
+
+	type args struct {
+		name string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{
+			name: "Should throw error when exceeds 249 characters",
+			args: args{
+				name: "verylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfsverylongkafkanamef8d9dkf9dkc11dfs",
+			},
+			wantErr: true,
+		},
+		{
+			name: "Should be valid when name is exactly 32 characters",
+			args: args{
+				name: "verylongtopicnamef8d9dkf9dkc11dd",
+			},
+			wantErr: false,
+		},
+		{
+			name: "Should be valid when containing dots",
+			args: args{
+				name: "k.afk.a",
+			},
+			wantErr: false,
+		},
+		{
+			name: "Should be invalid when containing special characters",
+			args: args{
+				name: "r*maujj??",
+			},
+			wantErr: false,
+		},
+		{
+			name: "Should be invalid when name is exactly 0 characters",
+			args: args{
+				name: "",
+			},
+			wantErr: true,
+		},
+		{
+			name: "Should be valid when using hyphens",
+			args: args{
+				name: "my-kafka-topic",
+			},
+			wantErr: false,
+		},
+		{
+			name: "Should be valid when starts with number",
+			args: args{
+				name: "1my-kafka-instance",
+			},
+			wantErr: false,
+		},
+		{
+			name: "Should be invalid when only two dots",
+			args: args{
+				name: "..",
+			},
+			wantErr: true,
+		},
+		{
+			name: "Should be invalid when only one dot",
+			args: args{
+				name: ".",
+			},
+			wantErr: true,
+		},
+		{
+			name: "Should be valid when only three dots",
+			args: args{
+				name: "...",
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// nolint
+			if err := ValidateName(tt.args.name); (err != nil) != tt.wantErr {
+				t.Errorf("ValidateName() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}