pro/im_manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: normalize kafka version config

Browse Source
This commit is contained in:
richjohnrambo
2026-08-28 15:36:56 +08:00
parent a938c6cb23
commit ed4f186d8e
4 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/queue/kafkamq.go
+10 -2
View File
@@ -160,7 +160,7 @@ func (r *KafkaMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg))
// RegisterKafkaMqConsumer 注册消费者
func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error) {
mqIns := &KafkaMq{}
kfkVersion, err := sarama.ParseKafkaVersion(connOpt.Version)
kfkVersion, err := parseKafkaVersion(connOpt.Version)
if err != nil {
return
}
@@ -203,7 +203,7 @@ func RegisterKafkaProducer(connOpt KafkaConfig) (client MqProducer, err error) {
// doRegisterKafkaProducer 注册同步类型实例
func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
kfkVersion, err := sarama.ParseKafkaVersion(connOpt.Version)
kfkVersion, err := parseKafkaVersion(connOpt.Version)
if err != nil {
return
}
@@ -239,6 +239,14 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
return
}
func parseKafkaVersion(version string) (sarama.KafkaVersion, error) {
parts := strings.Split(version, ".")
if len(parts) == 4 && parts[0] != "0" && parts[3] == "0" {
version = strings.Join(parts[:3], ".")
}
return sarama.ParseKafkaVersion(version)
}
func configureKafkaSASL(conf *sarama.Config, connOpt KafkaConfig) {
if connOpt.UserName == "" {
return