fix: enable local kafka docker integration
This commit is contained in:
@@ -118,8 +118,8 @@
|
||||
},
|
||||
|
||||
"kafka": {
|
||||
"enabled": true,
|
||||
"brokers": ["64.118.137.182:9094"],
|
||||
"enabled": $KAFKA_ENABLED,
|
||||
"brokers": $KAFKA_BROKERS_JSON,
|
||||
"client_id": "$KAFKA_CLIENT_ID",
|
||||
"consumer_group": "$KAFKA_CONSUMER_GROUP",
|
||||
"dial_timeout_ms": $KAFKA_DIAL_TIMEOUT_MS,
|
||||
@@ -127,11 +127,11 @@
|
||||
"read_timeout_ms": $KAFKA_READ_TIMEOUT_MS,
|
||||
"required_acks": $KAFKA_REQUIRED_ACKS,
|
||||
"allow_auto_topic_creation": $KAFKA_ALLOW_AUTO_TOPIC_CREATION,
|
||||
"tls_enabled": false,
|
||||
"sasl_enabled": true,
|
||||
"sasl_mechanism": "SCRAM-SHA-512",
|
||||
"sasl_username": "kafka-external",
|
||||
"sasl_password": "gr58tNnlZOVGW3AmMbL2",
|
||||
"tls_enabled": $KAFKA_TLS_ENABLED,
|
||||
"sasl_enabled": $KAFKA_SASL_ENABLED,
|
||||
"sasl_mechanism": "$KAFKA_SASL_MECHANISM",
|
||||
"sasl_username": "$KAFKA_SASL_USERNAME",
|
||||
"sasl_password": "$KAFKA_SASL_PASSWORD",
|
||||
"max_retry_count": $KAFKA_MAX_RETRY_COUNT,
|
||||
"retry_backoff_ms": $KAFKA_RETRY_BACKOFF_MS,
|
||||
"dead_letter_retention_seconds": $KAFKA_DEAD_LETTER_RETENTION_SECONDS,
|
||||
|
||||
@@ -274,9 +274,13 @@ func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
|
||||
if topic == "" {
|
||||
topic = topicName
|
||||
}
|
||||
groupID := kafkaBus.config.ConsumerGroup
|
||||
if topicName != "" {
|
||||
groupID += "." + topicName
|
||||
}
|
||||
reader := kafka.NewReader(kafka.ReaderConfig{
|
||||
Brokers: kafkaBus.config.Brokers,
|
||||
GroupID: kafkaBus.config.ConsumerGroup,
|
||||
GroupID: groupID,
|
||||
Topic: topic,
|
||||
StartOffset: kafka.FirstOffset,
|
||||
MinBytes: 1,
|
||||
@@ -312,7 +316,7 @@ func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
|
||||
}
|
||||
}
|
||||
}()
|
||||
logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, kafkaBus.config.ConsumerGroup)
|
||||
logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID)
|
||||
}
|
||||
|
||||
func (rt *kafkaRuntime) dialer() *kafka.Dialer {
|
||||
|
||||
Reference in New Issue
Block a user