update config and kafka init

internal/initialize/config.go

@@ -12,7 +12,8 @@ type Config struct {
 	HTTPPort            string `env:"HTTP_PORT" envDefault:"8001"`
 	PostgresCredentials string `env:"PG_CRED" envDefault:"host=localhost port=35432 user=squiz password=Redalert2 dbname=squiz sslmode=disable"`
 	KafkaBrokers        string `env:"KAFKA_BROKERS" envDefault:"localhost:9092"`
-	KafkaTopic          string `env:"KAFKA_TOPIC_TARIFF" envDefault:"test-topic"`
+	KafkaTopic          string `env:"KAFKA_TOPIC" envDefault:"test-topic"`
+	KafkaGroup          string `env:"KAFKA_GROUP" envDefault:"amoCRM"`
 	// урл в соц аус сервисе для генерации ссылки для авторизации в амо
 	PenaSocialAuthURL string `env:"PENA_SOCIAL_AUTH_URL" envDefault:"http://localhost:8000/amocrm/auth"`
 	// урл на который будет возвращен пользователь после авторизации это webhook/create get

internal/initialize/kafka.go

@@ -9,7 +9,7 @@ import (
 func KafkaConsumerInit(ctx context.Context, config Config) (*kgo.Client, error) {
 	kafkaClient, err := kgo.NewClient(
 		kgo.SeedBrokers(config.KafkaBrokers),
-		kgo.ConsumerGroup("amoCRM"),
+		kgo.ConsumerGroup(config.KafkaGroup),
 		kgo.DefaultProduceTopic(config.KafkaTopic),
 		kgo.ConsumeTopics(config.KafkaTopic),
 		kgo.ConsumeResetOffset(kgo.NewOffset().AfterMilli(time.Now().UnixMilli())),
@@ -24,5 +24,4 @@ func KafkaConsumerInit(ctx context.Context, config Config) (*kgo.Client, error)
 	}
 
 	return kafkaClient, nil
-
 }