*

.vscode/launch.json

@@ -13,6 +13,7 @@
       "env": {
         "APP_PORT":"9995",
         "KAFKA_PORT":"9092",
+        "KAFKA_HOST":"37.143.12.169",
       },
       "args": []
     }

internal/transport/kafka/writer.go

@@ -3,8 +3,6 @@ package kafka
 import (
 	"context"
 	"log"
-	"net"
-	"strconv"
 	"time"
 
 	"github.com/segmentio/kafka-go"
@@ -46,18 +44,18 @@ func (s *KafkaWriter) createTopic() error {
 	}
 	defer conn.Close()
 
-	//
+	// //
-	controller, era := conn.Controller()
+	// controller, era := conn.Controller()
-	if era != nil {
+	// if era != nil {
-		return era
+	// 	return era
-	}
+	// }
 
-	//
+	// //
-	controllerConn, eru := kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
+	// controllerConn, eru := kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))
-	if eru != nil {
+	// if eru != nil {
-		return eru
+	// 	return eru
-	}
+	// }
-	defer controllerConn.Close()
+	// defer controllerConn.Close()
 
 	//
 	topicConfigs := []kafka.TopicConfig{
@@ -68,7 +66,7 @@ func (s *KafkaWriter) createTopic() error {
 		},
 	}
 
-	return controllerConn.CreateTopics(topicConfigs...)
+	return conn.CreateTopics(topicConfigs...)
 }
 
 func (s *KafkaWriter) checkTopic() error {
@@ -99,6 +97,7 @@ func NewWriter(ctx context.Context, topic string, address ...string) *KafkaWrite
 			Topic:           topic,
 			Balancer:        &kafka.LeastBytes{},
 			WriteBackoffMax: time.Millisecond * 100,
+			BatchTimeout:    time.Millisecond * 100,
 			Addr:            kafka.TCP(address...),
 		},
 		first: address[0],

scripts/start.sh

@@ -3,5 +3,6 @@
 export APP_PORT=9995
 export KAFKA_PORT=9092
 export KAFKA_HOST=127.0.0.1
+export KAFKA_HOST=37.143.12.169
 
-./authDBService
+./authDBService