diff --git a/Dockerfile b/Dockerfile index d75b516..037d30c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -38,7 +38,7 @@ WORKDIR /app ENV QUEUE_DRIVER=kafka ENV KAFKA_ADDRESS=64.118.137.182:9094 -ENV KAFKA_VERSION=2.4.0.0 +ENV KAFKA_VERSION=2.4.0 ENV KAFKA_USERNAME=kafka-external ENV KAFKA_PASSWORD=gr58tNnlZOVGW3AmMbL2 ENV KAFKA_SASL_MECHANISM=SCRAM-SHA-512 diff --git a/server/internal/library/queue/kafkamq.go b/server/internal/library/queue/kafkamq.go index f115d2c..66e3738 100644 --- a/server/internal/library/queue/kafkamq.go +++ b/server/internal/library/queue/kafkamq.go @@ -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 diff --git a/server/manifest/config/config.example.yaml b/server/manifest/config/config.example.yaml index 35f1fc2..ae0461e 100644 --- a/server/manifest/config/config.example.yaml +++ b/server/manifest/config/config.example.yaml @@ -213,7 +213,7 @@ queue: logLevel: "info" # rocketmq日志输出级别,可选:all|close|debug|info|warn|error|fatal kafka: address: ["127.0.0.1:9092"] # kafka地址+端口 - version: "2.4.0.0" # kafka专属配置,默认2.4.0.0 + version: "2.4.0" # kafka专属配置,默认2.4.0 username: "" # SASL用户名,未开启认证时留空 password: "" # SASL密码,未开启认证时留空 mechanism: "SCRAM-SHA-512" # SASL机制,可选:SCRAM-SHA-512|SCRAM-SHA-256 diff --git a/server/manifest/docker/Dockerfile b/server/manifest/docker/Dockerfile index 936f8c7..4ce87d8 100644 --- a/server/manifest/docker/Dockerfile +++ b/server/manifest/docker/Dockerfile @@ -7,7 +7,7 @@ FROM loads/alpine:3.8 ENV WORKDIR /app ENV QUEUE_DRIVER kafka ENV KAFKA_ADDRESS 64.118.137.182:9094 -ENV KAFKA_VERSION 2.4.0.0 +ENV KAFKA_VERSION 2.4.0 ENV KAFKA_USERNAME kafka-external ENV KAFKA_PASSWORD gr58tNnlZOVGW3AmMbL2 ENV KAFKA_SASL_MECHANISM SCRAM-SHA-512