diff --git a/internal/kafka/writer.go b/internal/kafka/writer.go index 09276e4..58832b6 100644 --- a/internal/kafka/writer.go +++ b/internal/kafka/writer.go @@ -9,6 +9,7 @@ import ( "go.uber.org/zap" "google.golang.org/protobuf/proto" "message-handler/internal/config" + "time" ) const writeTopic = "message-handler" @@ -24,12 +25,13 @@ type kafkaNotifier struct { func NewKafkaNotifier(cfg *config.KafkaConfig, logger *zap.SugaredLogger) Notifier { w := &kafka.Writer{ - Addr: kafka.TCP(fmt.Sprintf("%s:%d", cfg.Host, cfg.Port)), - Topic: writeTopic, - Balancer: &kafka.LeastBytes{}, - Async: true, - BatchSize: 1, - ErrorLogger: kafka.LoggerFunc(logger.Errorw), + Addr: kafka.TCP(fmt.Sprintf("%s:%d", cfg.Host, cfg.Port)), + Topic: writeTopic, + Balancer: &kafka.LeastBytes{}, + Async: true, + BatchSize: 100, + BatchTimeout: 100 * time.Millisecond, + ErrorLogger: kafka.LoggerFunc(logger.Errorw), } return &kafkaNotifier{w: w}