From e0c35e4e9b086770da7a02ed6f158c9610d6bee6 Mon Sep 17 00:00:00 2001 From: Yvan <8574526@qq,com> Date: Sun, 30 Aug 2026 01:08:10 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go.mod | 2 + go.sum | 4 + internal/README.md | 2 +- .../biz/integration/integration_config.go | 75 ++- .../integration_config_communication_test.go | 67 +- .../integration_config_definition.go | 4 +- .../integration_config_runtime_test.go | 90 +++ internal/data/integration/migrations.go | 95 ++- internal/data/migrations_test.go | 8 +- internal/integration/connectivity.go | 2 +- .../connectivity_kafka_integration_test.go | 94 +++ internal/integration/connectivity_test.go | 11 + internal/integration/mq/emqx.go | 11 +- internal/integration/mq/emqx_test.go | 67 ++ internal/modules/catalog_test.go | 4 +- .../modules/integration/definition_test.go | 2 +- internal/server/httpx/response_test.go | 8 + pkg/mq/kafka.go | 592 ++++++++++++++++++ pkg/mq/kafka_integration_test.go | 116 ++++ pkg/mq/kafka_test.go | 234 +++++++ pkg/mq/subscription.go | 3 + pkg/mq/subscription_test.go | 54 +- .../view/systemTools/integration/config.vue | 6 +- 23 files changed, 1513 insertions(+), 38 deletions(-) create mode 100644 internal/integration/connectivity_kafka_integration_test.go create mode 100644 pkg/mq/kafka.go create mode 100644 pkg/mq/kafka_integration_test.go create mode 100644 pkg/mq/kafka_test.go diff --git a/go.mod b/go.mod index 1992898..c5b7060 100644 --- a/go.mod +++ b/go.mod @@ -32,6 +32,7 @@ require ( github.com/rabbitmq/amqp091-go v1.14.0 github.com/redis/go-redis/v9 v9.7.0 github.com/robfig/cron/v3 v3.0.1 + github.com/segmentio/kafka-go v0.4.51 github.com/shirou/gopsutil/v4 v4.25.7 github.com/spf13/viper v1.21.0 github.com/swaggo/files v1.0.1 @@ -143,6 +144,7 @@ require ( github.com/mozillazg/go-httpheader v0.2.1 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect + github.com/pierrec/lz4/v4 v4.1.15 // indirect github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/richardlehane/mscfb v1.0.4 // indirect diff --git a/go.sum b/go.sum index 7bd05d3..5256b8c 100644 --- a/go.sum +++ b/go.sum @@ -319,6 +319,8 @@ github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c h1:dAMKvw0MlJT1GshSTtih8C2gDs04w8dReiOGXrGLNoY= github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= +github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0= +github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8/go.mod h1:HKlIX3XHQyzLZPlr7++PzdhaXEj94dEiJgZDTsxEqUI= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= @@ -357,6 +359,8 @@ github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU= github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik= +github.com/segmentio/kafka-go v0.4.51 h1:JgDPPG75tC1rWIS2Me6MwcvXJ6f49UQ4HjAOef71Hno= +github.com/segmentio/kafka-go v0.4.51/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E= github.com/shirou/gopsutil/v4 v4.25.7 h1:bNb2JuqKuAu3tRlPv5piSmBZyMfecwQ+t/ILq+1JqVM= github.com/shirou/gopsutil/v4 v4.25.7/go.mod h1:XV/egmwJtd3ZQjBpJVY5kndsiOO4IRqy9TQnmm6VP7U= github.com/sijms/go-ora/v2 v2.7.17 h1:M/pYIqjaMUeBxyzOWp2oj4ntF6fHSBloJWGNH9vbmsU= diff --git a/internal/README.md b/internal/README.md index 2136eb7..4b6f0d1 100644 --- a/internal/README.md +++ b/internal/README.md @@ -11,7 +11,7 @@ - `config`：Viper 配置模型、加载、快照和热更新 - `data`：共享数据库生命周期；仓储按 `data/system`、`data/integration`、`data/task`、`data/payment` 隔离 - `initialize`：数据库首次初始化和系统种子数据编排 -- `integration`：Redis、邮件、对象存储、支付、WebSocket、EMQX 和 RabbitMQ 适配器 +- `integration`：Redis、邮件、对象存储、支付、WebSocket、EMQX、Kafka 和 RabbitMQ 适配器 - `routecatalog`：统一声明 HTTP 路由的公开性、操作审计、请求体策略和 API 元数据 - `server`：Gin server 组合与生命周期；横切 HTTP 代码按子包维护： `server/handler`、`server/middleware`、`server/router`、`server/staticfiles`； diff --git a/internal/biz/integration/integration_config.go b/internal/biz/integration/integration_config.go index 537f7bf..c1a4acd 100644 --- a/internal/biz/integration/integration_config.go +++ b/internal/biz/integration/integration_config.go @@ -79,6 +79,25 @@ type IntegrationConnectionTester interface { TestIntegration(context.Context, *IntegrationConfig) error } +type ConnectionTestError struct { + Provider string + Err error +} + +func (e *ConnectionTestError) Error() string { + if e == nil || strings.TrimSpace(e.Provider) == "" { + return "连接测试失败" + } + return strings.TrimSpace(e.Provider) + " 连接测试失败" +} + +func (e *ConnectionTestError) Unwrap() error { + if e == nil { + return nil + } + return e.Err +} + type IntegrationConfigUsecase struct { repo IntegrationConfigRepo tester IntegrationConnectionTester @@ -165,7 +184,14 @@ func (uc *IntegrationConfigUsecase) Test(ctx context.Context, config *Integratio if uc.tester == nil { return errors.New("集成连接测试器未初始化") } - return uc.tester.TestIntegration(ctx, config) + if err := uc.tester.TestIntegration(ctx, config); err != nil { + provider := config.Provider + if definition, ok := IntegrationDefinition(config.Kind, config.Provider); ok && strings.TrimSpace(definition.Name) != "" { + provider = definition.Name + } + return &ConnectionTestError{Provider: provider, Err: err} + } + return nil } func (uc *IntegrationConfigUsecase) Delete(ctx context.Context, kind, provider string) error { @@ -481,11 +507,52 @@ func IsIntegrationSecretKey(key string) bool { } func integrationInt64(values map[string]any, key string, fallback int64) int64 { - value := integrationText(values, key) - if value == "" { + value, exists := values[key] + if !exists || value == nil { return fallback } - parsed, err := strconv.ParseInt(value, 10, 64) + switch typed := value.(type) { + case int: + return int64(typed) + case int8: + return int64(typed) + case int16: + return int64(typed) + case int32: + return int64(typed) + case int64: + return typed + case uint: + return parseIntegrationInt64(strconv.FormatUint(uint64(typed), 10), fallback) + case uint8: + return int64(typed) + case uint16: + return int64(typed) + case uint32: + return int64(typed) + case uint64: + return parseIntegrationInt64(strconv.FormatUint(typed, 10), fallback) + case float32: + return parseIntegrationInt64(strconv.FormatFloat(float64(typed), 'f', -1, 32), fallback) + case float64: + return parseIntegrationInt64(strconv.FormatFloat(typed, 'f', -1, 64), fallback) + case json.Number: + if parsed, err := typed.Int64(); err == nil { + return parsed + } + if parsed, err := typed.Float64(); err == nil { + return parseIntegrationInt64(strconv.FormatFloat(parsed, 'f', -1, 64), fallback) + } + return fallback + case string: + return parseIntegrationInt64(typed, fallback) + default: + return parseIntegrationInt64(fmt.Sprint(typed), fallback) + } +} + +func parseIntegrationInt64(value string, fallback int64) int64 { + parsed, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64) if err != nil { return fallback } diff --git a/internal/biz/integration/integration_config_communication_test.go b/internal/biz/integration/integration_config_communication_test.go index 96c138c..072082b 100644 --- a/internal/biz/integration/integration_config_communication_test.go +++ b/internal/biz/integration/integration_config_communication_test.go @@ -3,6 +3,7 @@ package integration import ( "context" "encoding/json" + "errors" "testing" ) @@ -27,17 +28,19 @@ func (*integrationConfigRepoTestDouble) DeleteIntegrationConfig(context.Context, type integrationConnectionTesterDouble struct { calls int config *IntegrationConfig + err error } func (t *integrationConnectionTesterDouble) TestIntegration(_ context.Context, config *IntegrationConfig) error { t.calls++ t.config = config - return nil + return t.err } func TestCommunicationIntegrationDefinitionsAndValidation(t *testing.T) { for _, target := range []struct{ kind, provider string }{ {IntegrationKindMQ, "emqx"}, + {IntegrationKindMQ, "kafka"}, {IntegrationKindMQ, "rabbitmq"}, {IntegrationKindWebSocket, "melody"}, } { @@ -51,12 +54,49 @@ func TestCommunicationIntegrationDefinitionsAndValidation(t *testing.T) { } } +func TestKafkaDefaultConfigSurvivesJSONRoundTrip(t *testing.T) { + raw, err := json.Marshal(DefaultIntegrationConfig(IntegrationKindMQ, "kafka")) + if err != nil { + t.Fatal(err) + } + values, err := decodeIntegrationObject(raw) + if err != nil { + t.Fatal(err) + } + if got := integrationInt64(values, "max_bytes", 0); got != 10485760 { + t.Fatalf("max_bytes = %d, want 10485760; decoded=%#v", got, values["max_bytes"]) + } + if err = ValidateIntegrationConfig(IntegrationKindMQ, "kafka", values); err != nil { + t.Fatalf("JSON-decoded kafka defaults are invalid: %v", err) + } +} + +func TestIntegrationInt64RejectsFractionalAndOverflowValues(t *testing.T) { + values := map[string]any{"fractional": 1.5, "overflow": json.Number("9223372036854775808")} + if got := integrationInt64(values, "fractional", -1); got != -1 { + t.Fatalf("fractional value = %d, want fallback", got) + } + if got := integrationInt64(values, "overflow", -1); got != -1 { + t.Fatalf("overflow value = %d, want fallback", got) + } +} + func TestCommunicationIntegrationValidationRejectsInvalidValues(t *testing.T) { rabbit := DefaultIntegrationConfig(IntegrationKindMQ, "rabbitmq") rabbit["port"] = 0 if err := ValidateIntegrationConfig(IntegrationKindMQ, "rabbitmq", rabbit); err == nil { t.Fatal("invalid rabbitmq port was accepted") } + kafka := DefaultIntegrationConfig(IntegrationKindMQ, "kafka") + kafka["brokers"] = []string{"missing-port"} + if err := ValidateIntegrationConfig(IntegrationKindMQ, "kafka", kafka); err == nil { + t.Fatal("invalid kafka broker was accepted") + } + kafka = DefaultIntegrationConfig(IntegrationKindMQ, "kafka") + kafka["max_bytes"] = 0 + if err := ValidateIntegrationConfig(IntegrationKindMQ, "kafka", kafka); err == nil { + t.Fatal("invalid kafka byte limits were accepted") + } websocket := DefaultIntegrationConfig(IntegrationKindWebSocket, "melody") websocket["path"] = "ws" if err := ValidateIntegrationConfig(IntegrationKindWebSocket, "melody", websocket); err == nil { @@ -97,6 +137,31 @@ func TestIntegrationConfigTestDoesNotPersistCandidate(t *testing.T) { } } +func TestIntegrationConfigTestReturnsSafeProviderFailure(t *testing.T) { + underlying := errors.New("dial tcp 127.0.0.1:9092: connection refused") + for _, test := range []struct { + provider string + name string + }{ + {provider: "kafka", name: "Kafka"}, + {provider: "emqx", name: "EMQX"}, + {provider: "rabbitmq", name: "RabbitMQ"}, + } { + t.Run(test.provider, func(t *testing.T) { + values := DefaultIntegrationConfig(IntegrationKindMQ, test.provider) + raw, _ := json.Marshal(values) + usecase := NewIntegrationConfigUsecase(&integrationConfigRepoTestDouble{}, &integrationConnectionTesterDouble{err: underlying}) + err := usecase.Test(context.Background(), &IntegrationConfig{Kind: IntegrationKindMQ, Provider: test.provider, Values: raw}) + if err == nil || err.Error() != test.name+" 连接测试失败" { + t.Fatalf("Test() error = %v", err) + } + if !errors.Is(err, underlying) { + t.Fatalf("Test() error does not wrap the provider failure: %v", err) + } + }) + } +} + func TestIntegrationConfigRejectsJSONNull(t *testing.T) { repo := &integrationConfigRepoTestDouble{} usecase := NewIntegrationConfigUsecase(repo, &integrationConnectionTesterDouble{}) diff --git a/internal/biz/integration/integration_config_definition.go b/internal/biz/integration/integration_config_definition.go index 59c226a..26f9ab9 100644 --- a/internal/biz/integration/integration_config_definition.go +++ b/internal/biz/integration/integration_config_definition.go @@ -117,7 +117,7 @@ var integrationDefinitions = map[string][]IntegrationConfigDefinition{ {Key: "brokers", Label: "Broker 地址", Type: "string-list", Required: true, Placeholder: "127.0.0.1:9092", Description: "每行一个 host:port 地址。"}, {Key: "client_id", Label: "客户端 ID", Type: "text", Required: true, Placeholder: "kra"}, {Key: "group_id", Label: "消费组 ID", Type: "text", Required: true, Placeholder: "kra"}, - {Key: "username", Label: "SASL 用户名", Type: "text"}, + {Key: "username", Label: "SASL 用户名", Type: "text", Description: "当前使用 SASL/PLAIN；请与密码同时配置。"}, {Key: "password", Label: "SASL 密码", Type: "password", Secret: true}, {Key: "tls", Label: "启用 TLS", Type: "switch"}, {Key: "tls_skip_verify", Label: "跳过 TLS 证书校验", Type: "switch", Description: "仅用于受控测试环境。"}, @@ -127,7 +127,7 @@ var integrationDefinitions = map[string][]IntegrationConfigDefinition{ {Key: "max_wait", Label: "最大拉取等待（秒）", Type: "number", Required: true}, {Key: "connect_timeout", Label: "连接超时（秒）", Type: "number", Required: true}, {Key: "reconnect_interval", Label: "重连间隔（秒）", Type: "number", Required: true}, - {Key: "allow_auto_topic_creation", Label: "允许自动创建 Topic", Type: "switch"}, + {Key: "allow_auto_topic_creation", Label: "允许自动创建 Topic", Type: "switch", Description: "仅在生产消息时允许创建不存在的 Topic；订阅不会自动创建。"}, }, }, { diff --git a/internal/data/integration/integration_config_runtime_test.go b/internal/data/integration/integration_config_runtime_test.go index e45f943..49727e1 100644 --- a/internal/data/integration/integration_config_runtime_test.go +++ b/internal/data/integration/integration_config_runtime_test.go @@ -51,3 +51,93 @@ func TestIntegrationConfigSavePublishesUnmaskedRuntimeValues(t *testing.T) { t.Fatalf("runtime password = %#v", stored["password"]) } } + +func TestCommunicationDefaultsIncludeKafka(t *testing.T) { + db, err := openWithDriver("sqlite", "file:"+t.Name()+"?mode=memory&cache=shared") + if err != nil { + t.Fatal(err) + } + if err = db.AutoMigrate(&ConfigPO{}); err != nil { + t.Fatal(err) + } + if err = ensureCommunicationIntegrationConfigs(db); err != nil { + t.Fatal(err) + } + var row ConfigPO + if err = db.Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindMQ, "kafka").First(&row).Error; err != nil { + t.Fatal(err) + } + if row.Enabled || row.Config == "" { + t.Fatalf("kafka default row = %#v", row) + } +} + +func TestKafkaDefaultMigrationDoesNotRecreateOtherProviders(t *testing.T) { + db, err := openWithDriver("sqlite", "file:"+t.Name()+"?mode=memory&cache=shared") + if err != nil { + t.Fatal(err) + } + if err = db.AutoMigrate(&ConfigPO{}); err != nil { + t.Fatal(err) + } + if err = ensureKafkaIntegrationConfig(db); err != nil { + t.Fatal(err) + } + var kafkaRows, otherRows int64 + if err = db.Model(&ConfigPO{}).Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindMQ, "kafka").Count(&kafkaRows).Error; err != nil { + t.Fatal(err) + } + if err = db.Model(&ConfigPO{}).Where("provider <> ?", "kafka").Count(&otherRows).Error; err != nil { + t.Fatal(err) + } + if kafkaRows != 1 || otherRows != 0 { + t.Fatalf("migration rows kafka=%d other=%d", kafkaRows, otherRows) + } +} + +func TestCommunicationDefaultUpgradeRepairsRequiredEmptyFields(t *testing.T) { + db, err := openWithDriver("sqlite", "file:"+t.Name()+"?mode=memory&cache=shared") + if err != nil { + t.Fatal(err) + } + if err = db.AutoMigrate(&ConfigPO{}); err != nil { + t.Fatal(err) + } + legacy := `{"enabled":false,"broker":"","client_id":"","username":"","password":"","keep_alive":30,"clean_session":true,"connect_timeout":10}` + if err = db.Create(&ConfigPO{Kind: integrationbiz.IntegrationKindMQ, Provider: "emqx", Config: legacy}).Error; err != nil { + t.Fatal(err) + } + if err = upgradeCommunicationIntegrationDefaults(db); err != nil { + t.Fatal(err) + } + var row ConfigPO + if err = db.Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindMQ, "emqx").First(&row).Error; err != nil { + t.Fatal(err) + } + values := map[string]any{} + if err = json.Unmarshal([]byte(row.Config), &values); err != nil { + t.Fatal(err) + } + if values["broker"] != "tcp://127.0.0.1:1883" || values["client_id"] != "kra" || values["reconnect_interval"] != float64(5) { + t.Fatalf("upgraded emqx values = %#v", values) + } + if values["username"] != "" || values["password"] != "" { + t.Fatalf("optional credentials were overwritten: %#v", values) + } + if _, exists := values["enabled"]; exists { + t.Fatalf("legacy enabled field remained: %#v", values) + } + if err = db.Delete(&row).Error; err != nil { + t.Fatal(err) + } + if err = upgradeCommunicationIntegrationDefaults(db); err != nil { + t.Fatal(err) + } + var count int64 + if err = db.Model(&ConfigPO{}).Where("kind = ? AND provider = ?", integrationbiz.IntegrationKindMQ, "emqx").Count(&count).Error; err != nil { + t.Fatal(err) + } + if count != 0 { + t.Fatalf("upgrade recreated deleted emqx row: %d", count) + } +} diff --git a/internal/data/integration/migrations.go b/internal/data/integration/migrations.go index 622c677..f7c6507 100644 --- a/internal/data/integration/migrations.go +++ b/internal/data/integration/migrations.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "reflect" + "strings" integrationbiz "kra/internal/biz/integration" "kra/pkg/database/migration" @@ -18,7 +19,8 @@ func Migrations() []migration.Step { }}, {ID: "202608210001_communication_integration_defaults", Migrate: ensureCommunicationIntegrationConfigs}, {ID: "202608200004_payment_defaults", Migrate: ensurePaymentIntegrationConfigs}, - {ID: "202608290004_kafka_integration_default", Migrate: ensureCommunicationIntegrationConfigs}, + {ID: "202608290004_kafka_integration_default", Migrate: ensureKafkaIntegrationConfig}, + {ID: "202608290005_communication_integration_default_upgrade", Migrate: upgradeCommunicationIntegrationDefaults}, } } @@ -30,25 +32,90 @@ func ensureCommunicationIntegrationConfigs(db *gorm.DB) error { {integrationbiz.IntegrationKindWebSocket, "melody"}, } for _, item := range defaults { - var row ConfigPO - err := db.Where("kind = ? AND provider = ?", item.kind, item.provider).First(&row).Error - if errors.Is(err, gorm.ErrRecordNotFound) { - values, marshalErr := json.Marshal(integrationbiz.DefaultIntegrationConfig(item.kind, item.provider)) - if marshalErr != nil { - return marshalErr - } - if err = db.Create(&ConfigPO{Kind: item.kind, Provider: item.provider, Enabled: false, Config: string(values)}).Error; err != nil { - return err - } - continue - } - if err != nil { + if err := ensureIntegrationConfig(db, item.kind, item.provider); err != nil { return err } } return nil } +func ensureKafkaIntegrationConfig(db *gorm.DB) error { + return ensureIntegrationConfig(db, integrationbiz.IntegrationKindMQ, "kafka") +} + +func ensureIntegrationConfig(db *gorm.DB, kind, provider string) error { + var row ConfigPO + err := db.Where("kind = ? AND provider = ?", kind, provider).First(&row).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + values, marshalErr := json.Marshal(integrationbiz.DefaultIntegrationConfig(kind, provider)) + if marshalErr != nil { + return marshalErr + } + return db.Create(&ConfigPO{Kind: kind, Provider: provider, Enabled: false, Config: string(values)}).Error + } + return err +} + +func upgradeCommunicationIntegrationDefaults(db *gorm.DB) error { + for _, kind := range []string{integrationbiz.IntegrationKindMQ, integrationbiz.IntegrationKindWebSocket} { + for _, definition := range integrationbiz.IntegrationDefinitions(kind) { + var row ConfigPO + err := db.Where("kind = ? AND provider = ?", kind, definition.Provider).First(&row).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + continue + } + if err != nil { + return err + } + values := integrationObject(json.RawMessage(row.Config)) + changed := false + for key, value := range definition.Defaults { + if _, exists := values[key]; !exists { + values[key] = value + changed = true + } + } + for _, field := range definition.Fields { + if !field.Required || !emptyIntegrationConfigValue(values[field.Key]) || emptyIntegrationConfigValue(definition.Defaults[field.Key]) { + continue + } + values[field.Key] = definition.Defaults[field.Key] + changed = true + } + if _, exists := values["enabled"]; exists { + delete(values, "enabled") + changed = true + } + if !changed { + continue + } + encoded, marshalErr := json.Marshal(values) + if marshalErr != nil { + return marshalErr + } + if err := db.Model(&row).Update("config", string(encoded)).Error; err != nil { + return err + } + } + } + return nil +} + +func emptyIntegrationConfigValue(value any) bool { + switch typed := value.(type) { + case nil: + return true + case string: + return strings.TrimSpace(typed) == "" + case []any: + return len(typed) == 0 + case []string: + return len(typed) == 0 + default: + return false + } +} + func ensurePaymentIntegrationConfigs(db *gorm.DB) error { for _, definition := range integrationbiz.IntegrationDefinitions(integrationbiz.IntegrationKindPayment) { provider := definition.Provider diff --git a/internal/data/migrations_test.go b/internal/data/migrations_test.go index 0c44ce8..7d33548 100644 --- a/internal/data/migrations_test.go +++ b/internal/data/migrations_test.go @@ -34,15 +34,15 @@ func TestMigrateAllRunsModuleSchemasWithoutBootstrapSeed(t *testing.T) { if err = db.Table(migration.TableName).Count(&versions).Error; err != nil { t.Fatal(err) } - if versions != 11 { - t.Fatalf("migration versions = %d, want 11", versions) + if versions != 13 { + t.Fatalf("migration versions = %d, want 13", versions) } var communicationRows []dataintegration.ConfigPO if err = db.Where("kind IN ?", []string{"mq", "websocket"}).Order("kind, provider").Find(&communicationRows).Error; err != nil { t.Fatal(err) } - if len(communicationRows) != 3 { - t.Fatalf("communication integration rows = %d, want 3", len(communicationRows)) + if len(communicationRows) != 4 { + t.Fatalf("communication integration rows = %d, want 4", len(communicationRows)) } var paymentRows int64 if err = db.Model(&dataintegration.ConfigPO{}).Where("kind = ?", "payment").Count(&paymentRows).Error; err != nil { diff --git a/internal/integration/connectivity.go b/internal/integration/connectivity.go index 5cd2b94..ae6af95 100644 --- a/internal/integration/connectivity.go +++ b/internal/integration/connectivity.go @@ -74,7 +74,7 @@ func (t *ConnectivityTester) restoreMaskedSecrets(kind, provider string, values } for key := range masked { value, _ := values[key].(string) - if !config.IsMaskedSecret(strings.TrimSpace(value)) { + if strings.TrimSpace(value) != config.MaskedSecret { continue } prior, _ := currentValues[key].(string) diff --git a/internal/integration/connectivity_kafka_integration_test.go b/internal/integration/connectivity_kafka_integration_test.go new file mode 100644 index 0000000..0616fbc --- /dev/null +++ b/internal/integration/connectivity_kafka_integration_test.go @@ -0,0 +1,94 @@ +package integration + +import ( + "context" + "encoding/json" + "fmt" + "net" + "os" + "strconv" + "strings" + "testing" + "time" + + integrationbiz "kra/internal/biz/integration" +) + +type kafkaConnectivityRepo struct{} + +func (*kafkaConnectivityRepo) ListIntegrationConfigs(context.Context, string) ([]*integrationbiz.IntegrationConfig, error) { + return nil, nil +} +func (*kafkaConnectivityRepo) FindIntegrationConfig(context.Context, string, string) (*integrationbiz.IntegrationConfig, error) { + return nil, nil +} +func (*kafkaConnectivityRepo) SaveIntegrationConfig(context.Context, *integrationbiz.IntegrationConfig) error { + return nil +} +func (*kafkaConnectivityRepo) DeleteIntegrationConfig(context.Context, string, string) error { + return nil +} + +func TestKafkaConnectivityIntegration(t *testing.T) { + rawBrokers := strings.TrimSpace(os.Getenv("KRA_KAFKA_TEST_BROKERS")) + if rawBrokers == "" { + t.Skip("KRA_KAFKA_TEST_BROKERS is not configured") + } + brokers := make([]string, 0) + for _, broker := range strings.Split(rawBrokers, ",") { + if broker = strings.TrimSpace(broker); broker != "" { + brokers = append(brokers, broker) + } + } + values := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindMQ, "kafka") + values["brokers"] = brokers + values["client_id"] = fmt.Sprintf("kra-connectivity-%d", time.Now().UnixNano()) + values["group_id"] = fmt.Sprintf("kra-connectivity-%d", time.Now().UnixNano()) + testMQConnectivityIntegration(t, "kafka", values) +} + +func TestEMQXConnectivityIntegration(t *testing.T) { + broker := strings.TrimSpace(os.Getenv("KRA_EMQX_TEST_BROKER")) + if broker == "" { + t.Skip("KRA_EMQX_TEST_BROKER is not configured") + } + values := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindMQ, "emqx") + values["broker"] = broker + values["client_id"] = fmt.Sprintf("kra-connectivity-%d", time.Now().UnixNano()) + testMQConnectivityIntegration(t, "emqx", values) +} + +func TestRabbitMQConnectivityIntegration(t *testing.T) { + address := strings.TrimSpace(os.Getenv("KRA_RABBITMQ_TEST_ADDR")) + if address == "" { + t.Skip("KRA_RABBITMQ_TEST_ADDR is not configured") + } + host, portText, err := net.SplitHostPort(address) + if err != nil { + t.Fatal(err) + } + port, err := strconv.Atoi(portText) + if err != nil { + t.Fatal(err) + } + values := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindMQ, "rabbitmq") + values["host"] = host + values["port"] = port + testMQConnectivityIntegration(t, "rabbitmq", values) +} + +func testMQConnectivityIntegration(t *testing.T, provider string, values map[string]any) { + t.Helper() + raw, err := json.Marshal(values) + if err != nil { + t.Fatal(err) + } + usecase := integrationbiz.NewIntegrationConfigUsecase(&kafkaConnectivityRepo{}, NewConnectivityTester(nil)) + if err = usecase.Test(context.Background(), &integrationbiz.IntegrationConfig{ + Kind: integrationbiz.IntegrationKindMQ, + Provider: provider, + Values: raw, + }); err != nil { + t.Fatalf("test JSON-decoded %s defaults: %v", provider, err) + } +} diff --git a/internal/integration/connectivity_test.go b/internal/integration/connectivity_test.go index 0528226..9e1d49f 100644 --- a/internal/integration/connectivity_test.go +++ b/internal/integration/connectivity_test.go @@ -46,6 +46,17 @@ func TestConnectivityTesterRestoreMaskedSecrets(t *testing.T) { } }) + t.Run("keeps an empty optional secret", func(t *testing.T) { + values := map[string]any{"password": ""} + + if err := tester.restoreMaskedSecrets("mq", "kafka", values); err != nil { + t.Fatalf("restoreMaskedSecrets() error = %v", err) + } + if got := values["password"]; got != "" { + t.Fatalf("password = %q, want empty value", got) + } + }) + t.Run("does not treat non-secret fields as masked secrets", func(t *testing.T) { values := map[string]any{ "username": "******", diff --git a/internal/integration/mq/emqx.go b/internal/integration/mq/emqx.go index 908ded2..cc0e7e3 100644 --- a/internal/integration/mq/emqx.go +++ b/internal/integration/mq/emqx.go @@ -279,7 +279,13 @@ func configStrings(values map[string]any, key string) []string { } stringsValue, ok := values[key].([]string) if ok { - return append([]string(nil), stringsValue...) + result := make([]string, 0, len(stringsValue)) + for _, item := range stringsValue { + if value := strings.TrimSpace(item); value != "" { + result = append(result, value) + } + } + return result } return nil } @@ -327,6 +333,9 @@ func (r *Reloadable) retryOnce() { client := r.clientLocked(provider) if client != nil && client.Connected() { if r.pending[provider] { + if retryAt := r.nextRetry[provider]; !retryAt.IsZero() && now.Before(retryAt) { + continue + } if err := r.reconcileProviderLocked(provider); err != nil { r.scheduleRetryLocked(provider, config.Values) } diff --git a/internal/integration/mq/emqx_test.go b/internal/integration/mq/emqx_test.go index b493c93..dbae291 100644 --- a/internal/integration/mq/emqx_test.go +++ b/internal/integration/mq/emqx_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "log/slog" + "strings" "testing" "time" @@ -209,3 +210,69 @@ func TestReloadableLegacySubscribeDoesNotLeaveOfflineDeclaration(t *testing.T) { t.Fatalf("offline legacy declaration remained: %#v", r.subscriptions[ProviderEMQX]) } } + +func TestReloadableExposesKafkaNamedClient(t *testing.T) { + r := &Reloadable{} + client := r.Client(" KAFKA ") + if client == nil { + t.Fatal("kafka named client is nil") + } + if client.Connected() { + t.Fatal("unconfigured kafka named client reported connected") + } +} + +func TestConfigStringsDecodesJSONArrays(t *testing.T) { + values := map[string]any{"brokers": []any{" kafka-1:9092 ", "", "kafka-2:9092"}} + brokers := configStrings(values, "brokers") + if len(brokers) != 2 || brokers[0] != "kafka-1:9092" || brokers[1] != "kafka-2:9092" { + t.Fatalf("brokers = %#v", brokers) + } +} + +func TestConfigStringsNormalizesTypedSlices(t *testing.T) { + values := map[string]any{"brokers": []string{" kafka-1:9092 ", "", "kafka-2:9092"}} + brokers := configStrings(values, "brokers") + if len(brokers) != 2 || brokers[0] != "kafka-1:9092" || brokers[1] != "kafka-2:9092" { + t.Fatalf("brokers = %#v", brokers) + } +} + +func TestReloadableRejectsUnsupportedKafkaQoS(t *testing.T) { + r := &Reloadable{} + err := r.Register(platformmq.SubscriptionSet{ + Owner: "orders", + Provider: ProviderKafka, + Topics: []platformmq.TopicSubscription{{Topic: "orders.created", QoS: platformmq.ExactlyOnce, Handler: func(context.Context, platformmq.Message) {}}}, + }) + if err == nil || !strings.Contains(err.Error(), "supports qos 0 or 1") { + t.Fatalf("Register() error = %v", err) + } +} + +func TestRetryOnceHonorsPendingSubscriptionDeadline(t *testing.T) { + client := &fakeClient{} + r := &Reloadable{ + clients: map[string]platformmq.Client{ProviderKafka: client}, + configs: map[string]runtimeconfig.Config{ + ProviderKafka: {Kind: "mq", Provider: ProviderKafka, Enabled: true}, + }, + subscriptions: map[string]map[string]map[string]subscription{ + ProviderKafka: { + "orders.created": {"orders": {qos: platformmq.AtLeastOnce, handler: func(context.Context, platformmq.Message) {}}}, + }, + }, + bindings: map[string]map[string]byte{ProviderKafka: {}}, + pending: map[string]bool{ProviderKafka: true}, + nextRetry: map[string]time.Time{ProviderKafka: time.Now().Add(time.Minute)}, + } + r.retryOnce() + if len(client.subscribed) != 0 { + t.Fatalf("subscription retried before deadline: %v", client.subscribed) + } + r.nextRetry[ProviderKafka] = time.Now().Add(-time.Second) + r.retryOnce() + if len(client.subscribed) != 1 || client.subscribed[0] != "orders.created" { + t.Fatalf("subscription was not retried after deadline: %v", client.subscribed) + } +} diff --git a/internal/modules/catalog_test.go b/internal/modules/catalog_test.go index ff06848..913aabc 100644 --- a/internal/modules/catalog_test.go +++ b/internal/modules/catalog_test.go @@ -25,8 +25,8 @@ func TestCatalogContainsBuiltInModulesInDependencyOrder(t *testing.T) { t.Fatalf("definition[%d] = %q, want %q", index, got, name) } } - if got := catalog.MigrationSteps(); len(got) != 11 { - t.Fatalf("module migrations = %d, want 11", len(got)) + if got := catalog.MigrationSteps(); len(got) != 13 { + t.Fatalf("module migrations = %d, want 13", len(got)) } var menus, apis int for _, definition := range catalog.Definitions { diff --git a/internal/modules/integration/definition_test.go b/internal/modules/integration/definition_test.go index 3567a64..1eea53f 100644 --- a/internal/modules/integration/definition_test.go +++ b/internal/modules/integration/definition_test.go @@ -4,7 +4,7 @@ import "testing" func TestDefinitionOwnsCommunicationSurface(t *testing.T) { definition := Definition() - if definition.Name != "integration" || len(definition.Migrations) != 3 || len(definition.Surface.Menus) != 1 || len(definition.Surface.APIs) != 5 { + if definition.Name != "integration" || len(definition.Migrations) != 5 || len(definition.Surface.Menus) != 1 || len(definition.Surface.APIs) != 5 { t.Fatalf("integration surface = %#v", definition.Surface) } } diff --git a/internal/server/httpx/response_test.go b/internal/server/httpx/response_test.go index e3b03e2..0c0435e 100644 --- a/internal/server/httpx/response_test.go +++ b/internal/server/httpx/response_test.go @@ -33,6 +33,14 @@ func TestResponseHelpers(t *testing.T) { } } +func TestSanitizeFailureMessageKeepsSafeConnectionSummary(t *testing.T) { + for _, message := range []string{"Kafka 连接测试失败", "EMQX 连接测试失败", "RabbitMQ 连接测试失败"} { + if got := sanitizeFailureMessage(message); got != message { + t.Fatalf("sanitizeFailureMessage(%q) = %q", message, got) + } + } +} + func TestSetCookieUsesRequestedName(t *testing.T) { gin.SetMode(gin.TestMode) r := gin.New() diff --git a/pkg/mq/kafka.go b/pkg/mq/kafka.go new file mode 100644 index 0000000..ccdbf84 --- /dev/null +++ b/pkg/mq/kafka.go @@ -0,0 +1,592 @@ +package mq + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + kafkago "github.com/segmentio/kafka-go" + "github.com/segmentio/kafka-go/sasl" + "github.com/segmentio/kafka-go/sasl/plain" +) + +type KafkaConfig struct { + Enabled bool + Brokers []string + ClientID string + GroupID string + Username string + Password string + TLS bool + TLSSkipVerify bool + StartOffset string + MinBytes int + MaxBytes int + MaxWait time.Duration + ConnectTimeout time.Duration + ReconnectInterval time.Duration + AllowAutoTopicCreation bool +} + +type kafkaSubscription struct { + reader *kafkago.Reader + cancel context.CancelFunc + qos byte + handler Handler +} + +type kafkaTopicMetadataError struct { + err error +} + +func (e *kafkaTopicMetadataError) Error() string { return e.err.Error() } +func (e *kafkaTopicMetadataError) Unwrap() error { return e.err } + +// Kafka adapts Kafka topics and consumer groups to the shared messaging +// contract. QoS 0 commits before dispatch; QoS 1 commits after dispatch. +type Kafka struct { + opMu sync.Mutex + mu sync.RWMutex + config KafkaConfig + dialer *kafkago.Dialer + metadata *kafkago.Client + transport *kafkago.Transport + bestEffort *kafkago.Writer + acknowledged *kafkago.Writer + subscriptions map[string]*kafkaSubscription + connected atomic.Bool + reconnectAt atomic.Int64 + closed bool +} + +func NewKafka(config KafkaConfig) (*Kafka, error) { + client := &Kafka{subscriptions: make(map[string]*kafkaSubscription)} + if !config.Enabled { + return client, nil + } + config = defaultKafkaConfig(config) + if err := validateKafkaConfig(config); err != nil { + return nil, err + } + + tlsConfig, mechanism := kafkaSecurity(config) + dialer := &kafkago.Dialer{ + ClientID: config.ClientID, + Timeout: config.ConnectTimeout, + TLS: tlsConfig, + SASLMechanism: mechanism, + } + transport := &kafkago.Transport{ + DialTimeout: config.ConnectTimeout, + ClientID: config.ClientID, + TLS: tlsConfig, + SASL: mechanism, + } + metadata := &kafkago.Client{ + Addr: kafkago.TCP(config.Brokers...), + Timeout: config.ConnectTimeout, + Transport: transport, + } + if err := probeKafkaMetadata(context.Background(), metadata, ""); err != nil { + transport.CloseIdleConnections() + return nil, fmt.Errorf("connect kafka: %w", err) + } + newWriter := func(acks kafkago.RequiredAcks) *kafkago.Writer { + return &kafkago.Writer{ + Addr: kafkago.TCP(config.Brokers...), + Balancer: &kafkago.LeastBytes{}, + RequiredAcks: acks, + ReadTimeout: config.ConnectTimeout, + WriteTimeout: config.ConnectTimeout, + Transport: transport, + AllowAutoTopicCreation: config.AllowAutoTopicCreation, + } + } + client.config = config + client.dialer = dialer + client.metadata = metadata + client.transport = transport + client.bestEffort = newWriter(kafkago.RequireNone) + client.acknowledged = newWriter(kafkago.RequireAll) + client.markConnected() + return client, nil +} + +func probeKafkaMetadata(ctx context.Context, client *kafkago.Client, topic string) error { + if client == nil { + return ErrUnavailable + } + request := &kafkago.MetadataRequest{Topics: []string{}} + if topic != "" { + request.Topics = []string{topic} + } + response, err := client.Metadata(nonNilContext(ctx), request) + if err != nil { + return err + } + if response == nil || len(response.Brokers) == 0 { + return errors.New("kafka metadata contains no brokers") + } + if topic == "" { + return nil + } + for _, item := range response.Topics { + if item.Name != topic { + continue + } + if item.Error != nil { + return &kafkaTopicMetadataError{err: item.Error} + } + if len(item.Partitions) == 0 { + return &kafkaTopicMetadataError{err: errors.New("topic has no partitions")} + } + return nil + } + return &kafkaTopicMetadataError{err: errors.New("topic metadata was not returned")} +} + +func defaultKafkaConfig(config KafkaConfig) KafkaConfig { + brokers := make([]string, 0, len(config.Brokers)) + for _, broker := range config.Brokers { + if broker = strings.TrimSpace(broker); broker != "" { + brokers = append(brokers, broker) + } + } + config.Brokers = brokers + config.ClientID = strings.TrimSpace(config.ClientID) + config.GroupID = strings.TrimSpace(config.GroupID) + config.StartOffset = strings.ToLower(strings.TrimSpace(config.StartOffset)) + if config.StartOffset == "" { + config.StartOffset = "earliest" + } + if config.MinBytes <= 0 { + config.MinBytes = 1 + } + if config.MaxBytes <= 0 { + config.MaxBytes = 10 << 20 + } + if config.MaxWait <= 0 { + config.MaxWait = time.Second + } + if config.ConnectTimeout <= 0 { + config.ConnectTimeout = 10 * time.Second + } + if config.ReconnectInterval <= 0 { + config.ReconnectInterval = 5 * time.Second + } + return config +} + +func validateKafkaConfig(config KafkaConfig) error { + if len(config.Brokers) == 0 { + return errors.New("kafka brokers are empty") + } + for _, broker := range config.Brokers { + host, portText, err := net.SplitHostPort(broker) + port, parseErr := strconv.Atoi(portText) + if err != nil || parseErr != nil || strings.TrimSpace(host) == "" || port < 1 || port > 65535 { + return fmt.Errorf("invalid kafka broker %q", broker) + } + } + if config.ClientID == "" { + return errors.New("kafka client id is empty") + } + if config.GroupID == "" { + return errors.New("kafka group id is empty") + } + if (config.Username == "") != (config.Password == "") { + return errors.New("kafka username and password must be configured together") + } + if config.TLSSkipVerify && !config.TLS { + return errors.New("kafka tls skip verify requires tls") + } + if config.StartOffset != "earliest" && config.StartOffset != "latest" { + return fmt.Errorf("invalid kafka start offset %q", config.StartOffset) + } + if config.MinBytes < 1 || config.MaxBytes < config.MinBytes { + return errors.New("kafka byte limits are invalid") + } + return nil +} + +func kafkaSecurity(config KafkaConfig) (*tls.Config, sasl.Mechanism) { + var tlsConfig *tls.Config + if config.TLS { + tlsConfig = &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: config.TLSSkipVerify} //nolint:gosec // explicit operator setting + } + var mechanism sasl.Mechanism + if config.Username != "" { + mechanism = plain.Mechanism{Username: config.Username, Password: config.Password} + } + return tlsConfig, mechanism +} + +func (c *Kafka) Publish(ctx context.Context, topic string, payload []byte, qos byte, retain bool) error { + topic = strings.TrimSpace(topic) + if topic == "" { + return errors.New("kafka topic is empty") + } + if retain { + return errors.New("kafka does not support retained messages") + } + if qos > AtLeastOnce { + return fmt.Errorf("kafka supports qos 0 or 1, got %d", qos) + } + if c == nil { + return ErrUnavailable + } + ctx = nonNilContext(ctx) + c.mu.RLock() + if c.closed { + c.mu.RUnlock() + return ErrUnavailable + } + writer := c.bestEffort + if qos == AtLeastOnce { + writer = c.acknowledged + } + config := c.config + c.mu.RUnlock() + if writer == nil { + return ErrUnavailable + } + err := writeKafkaMessage(ctx, writer, kafkago.Message{Topic: topic, Value: append([]byte(nil), payload...)}, config) + if err == nil { + c.markConnected() + } + return err +} + +func writeKafkaMessage(ctx context.Context, writer *kafkago.Writer, message kafkago.Message, config KafkaConfig) error { + if writer == nil { + return ErrUnavailable + } + ctx = nonNilContext(ctx) + if !config.AllowAutoTopicCreation { + return writer.WriteMessages(ctx, message) + } + attemptCtx, cancel := context.WithTimeout(ctx, config.ConnectTimeout) + defer cancel() + delay := 100 * time.Millisecond + var lastErr error + for { + lastErr = writer.WriteMessages(attemptCtx, message) + if lastErr == nil || !kafkaTopicNotReady(lastErr) { + return lastErr + } + timer := time.NewTimer(delay) + select { + case <-attemptCtx.Done(): + timer.Stop() + if err := ctx.Err(); err != nil { + return err + } + return lastErr + case <-timer.C: + } + if delay < time.Second { + delay *= 2 + if delay > time.Second { + delay = time.Second + } + } + } +} + +func kafkaTopicNotReady(err error) bool { + if err == nil { + return false + } + if writeErrors, ok := err.(kafkago.WriteErrors); ok { + found := false + for _, item := range writeErrors { + if item == nil { + continue + } + found = true + if !kafkaTopicNotReady(item) { + return false + } + } + return found + } + var kafkaErr kafkago.Error + if !errors.As(err, &kafkaErr) { + return false + } + return kafkaErr == kafkago.UnknownTopicOrPartition || kafkaErr == kafkago.LeaderNotAvailable || kafkaErr == kafkago.NotLeaderForPartition +} + +func (c *Kafka) Subscribe(ctx context.Context, topic string, qos byte, handler Handler) error { + topic = strings.TrimSpace(topic) + if topic == "" { + return errors.New("kafka topic is empty") + } + if qos > AtLeastOnce { + return fmt.Errorf("kafka supports qos 0 or 1, got %d", qos) + } + if handler == nil { + return errors.New("kafka handler is nil") + } + if c == nil { + return ErrUnavailable + } + ctx = nonNilContext(ctx) + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + c.opMu.Lock() + defer c.opMu.Unlock() + c.mu.RLock() + if c.closed || c.dialer == nil || c.metadata == nil { + c.mu.RUnlock() + return ErrUnavailable + } + config := c.config + dialer := c.dialer + metadata := c.metadata + c.mu.RUnlock() + if err := probeKafkaMetadata(ctx, metadata, topic); err != nil { + var topicErr *kafkaTopicMetadataError + if errors.As(err, &topicErr) { + c.markConnected() + } else if ctx.Err() == nil { + c.markUnavailable() + } + return fmt.Errorf("lookup kafka topic %q: %w", topic, err) + } + if old := c.removeSubscription(topic); old != nil { + closeKafkaSubscription(old) + } + reader := kafkago.NewReader(kafkago.ReaderConfig{ + Brokers: append([]string(nil), config.Brokers...), + GroupID: config.GroupID, + Topic: topic, + Dialer: dialer, + MinBytes: config.MinBytes, + MaxBytes: config.MaxBytes, + MaxWait: config.MaxWait, + JoinGroupBackoff: config.ReconnectInterval, + ReadLagInterval: -1, + StartOffset: kafkaStartOffset(config.StartOffset), + }) + consumeCtx, cancel := context.WithCancel(context.Background()) + subscription := &kafkaSubscription{reader: reader, cancel: cancel, qos: qos, handler: handler} + c.mu.Lock() + if c.closed { + c.mu.Unlock() + cancel() + _ = reader.Close() + return ErrUnavailable + } + c.subscriptions[topic] = subscription + c.mu.Unlock() + c.markConnected() + go c.consume(consumeCtx, subscription) + return nil +} + +func kafkaStartOffset(value string) int64 { + if value == "latest" { + return kafkago.LastOffset + } + return kafkago.FirstOffset +} + +func (c *Kafka) consume(ctx context.Context, subscription *kafkaSubscription) { + for { + var message kafkago.Message + var err error + if subscription.qos == AtMostOnce { + message, err = subscription.reader.ReadMessage(ctx) + } else { + message, err = subscription.reader.FetchMessage(ctx) + } + if err != nil { + if ctx.Err() != nil || errors.Is(err, io.EOF) { + return + } + c.markUnavailable() + if !sleepContext(ctx, c.config.ReconnectInterval) { + return + } + continue + } + c.markConnected() + subscription.handler(ctx, Message{Topic: message.Topic, Payload: append([]byte(nil), message.Value...), QoS: subscription.qos}) + if ctx.Err() != nil { + return + } + if subscription.qos == AtLeastOnce { + if err = subscription.reader.CommitMessages(ctx, message); err != nil { + c.markUnavailable() + } + } + } +} + +func sleepContext(ctx context.Context, duration time.Duration) bool { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +func (c *Kafka) Unsubscribe(ctx context.Context, topics ...string) error { + if len(topics) == 0 { + return errors.New("kafka topics are empty") + } + if c == nil { + return ErrUnavailable + } + ctx = nonNilContext(ctx) + normalized := make([]string, len(topics)) + for index, topic := range topics { + normalized[index] = strings.TrimSpace(topic) + if normalized[index] == "" { + return errors.New("kafka topic is empty") + } + } + c.opMu.Lock() + defer c.opMu.Unlock() + for _, topic := range normalized { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + if subscription := c.removeSubscription(topic); subscription != nil { + closeKafkaSubscription(subscription) + } + } + return nil +} + +func (c *Kafka) removeSubscription(topic string) *kafkaSubscription { + c.mu.Lock() + defer c.mu.Unlock() + subscription := c.subscriptions[topic] + delete(c.subscriptions, topic) + return subscription +} + +func closeKafkaSubscription(subscription *kafkaSubscription) { + if subscription == nil { + return + } + if subscription.cancel != nil { + subscription.cancel() + } + if subscription.reader != nil { + _ = subscription.reader.Close() + } +} + +func (c *Kafka) Connected() bool { + if c == nil { + return false + } + c.mu.RLock() + closed := c.closed + c.mu.RUnlock() + return !closed && c.connected.Load() +} + +// Reconnecting lets the outer reloadable honor the configured retry delay +// before replacing a client after a runtime broker failure. +func (c *Kafka) Reconnecting() bool { + if c == nil { + return false + } + c.mu.RLock() + closed := c.closed + c.mu.RUnlock() + return !closed && time.Now().UnixNano() < c.reconnectAt.Load() +} + +func (c *Kafka) markConnected() { + if c == nil { + return + } + c.mu.RLock() + closed := c.closed + c.mu.RUnlock() + if closed { + return + } + c.reconnectAt.Store(0) + c.connected.Store(true) +} + +func (c *Kafka) markUnavailable() { + if c == nil { + return + } + c.mu.RLock() + closed := c.closed + c.mu.RUnlock() + if closed { + return + } + c.connected.Store(false) + c.reconnectAt.Store(time.Now().Add(c.config.ReconnectInterval).UnixNano()) +} + +func (c *Kafka) Close() error { + if c == nil { + return nil + } + c.opMu.Lock() + defer c.opMu.Unlock() + c.mu.Lock() + if c.closed { + c.mu.Unlock() + return nil + } + c.closed = true + c.connected.Store(false) + c.reconnectAt.Store(0) + subscriptions := make([]*kafkaSubscription, 0, len(c.subscriptions)) + for topic, subscription := range c.subscriptions { + subscriptions = append(subscriptions, subscription) + delete(c.subscriptions, topic) + } + bestEffort := c.bestEffort + acknowledged := c.acknowledged + transport := c.transport + c.bestEffort = nil + c.acknowledged = nil + c.transport = nil + c.dialer = nil + c.metadata = nil + c.mu.Unlock() + for _, subscription := range subscriptions { + closeKafkaSubscription(subscription) + } + var result error + if bestEffort != nil { + result = errors.Join(result, bestEffort.Close()) + } + if acknowledged != nil { + result = errors.Join(result, acknowledged.Close()) + } + if transport != nil { + transport.CloseIdleConnections() + } + return result +} diff --git a/pkg/mq/kafka_integration_test.go b/pkg/mq/kafka_integration_test.go new file mode 100644 index 0000000..671cd14 --- /dev/null +++ b/pkg/mq/kafka_integration_test.go @@ -0,0 +1,116 @@ +package mq + +import ( + "context" + "fmt" + "os" + "strings" + "testing" + "time" +) + +func TestKafkaRoundTripIntegration(t *testing.T) { + rawBrokers := strings.TrimSpace(os.Getenv("KRA_KAFKA_TEST_BROKERS")) + if rawBrokers == "" { + t.Skip("KRA_KAFKA_TEST_BROKERS is not configured") + } + brokers := []string{"127.0.0.1:1"} + for _, broker := range strings.Split(rawBrokers, ",") { + if broker = strings.TrimSpace(broker); broker != "" { + brokers = append(brokers, broker) + } + } + if len(brokers) == 1 { + t.Fatal("KRA_KAFKA_TEST_BROKERS contains no broker addresses") + } + + suffix := fmt.Sprintf("%d", time.Now().UnixNano()) + topics := []string{"kra-review-qos1-" + suffix, "kra-review-qos0-" + suffix} + groupID := "kra-review-" + suffix + newClient := func(clientID string) *Kafka { + client, err := NewKafka(KafkaConfig{ + Enabled: true, + Brokers: brokers, + ClientID: clientID, + GroupID: groupID, + StartOffset: "earliest", + MinBytes: 1, + MaxBytes: 1 << 20, + MaxWait: 100 * time.Millisecond, + ConnectTimeout: 5 * time.Second, + ReconnectInterval: 500 * time.Millisecond, + AllowAutoTopicCreation: true, + }) + if err != nil { + t.Fatalf("NewKafka(%s): %v", clientID, err) + } + return client + } + + client := newClient("kra-review-producer-" + suffix) + t.Cleanup(func() { _ = client.Close() }) + firstPayloads := [][]byte{[]byte("qos1-first"), []byte("qos0-first")} + qosValues := []byte{AtLeastOnce, AtMostOnce} + for index, topic := range topics { + if err := client.Publish(context.Background(), topic, firstPayloads[index], qosValues[index], false); err != nil { + t.Fatalf("publish first message to %s: %v", topic, err) + } + } + + firstReceived := []chan Message{make(chan Message, 1), make(chan Message, 1)} + for index, topic := range topics { + index := index + if err := client.Subscribe(context.Background(), topic, qosValues[index], func(_ context.Context, message Message) { + firstReceived[index] <- message + }); err != nil { + t.Fatalf("subscribe to %s: %v", topic, err) + } + } + for index, received := range firstReceived { + message := waitKafkaMessage(t, received, 30*time.Second) + if string(message.Payload) != string(firstPayloads[index]) || message.QoS != qosValues[index] { + t.Fatalf("first message[%d] = payload %q qos %d", index, message.Payload, message.QoS) + } + } + // QoS 1 commits after the handler returns. Give the synchronous commit time + // to finish before closing the readers and rejoining the same group. + time.Sleep(500 * time.Millisecond) + if err := client.Unsubscribe(context.Background(), topics...); err != nil { + t.Fatal(err) + } + + replayClient := newClient("kra-review-replay-" + suffix) + t.Cleanup(func() { _ = replayClient.Close() }) + replayed := []chan Message{make(chan Message, 2), make(chan Message, 2)} + for index, topic := range topics { + index := index + if err := replayClient.Subscribe(context.Background(), topic, qosValues[index], func(_ context.Context, message Message) { + replayed[index] <- message + }); err != nil { + t.Fatalf("replay subscribe to %s: %v", topic, err) + } + } + secondPayloads := [][]byte{[]byte("qos1-second"), []byte("qos0-second")} + for index, topic := range topics { + if err := replayClient.Publish(context.Background(), topic, secondPayloads[index], qosValues[index], false); err != nil { + t.Fatalf("publish second message to %s: %v", topic, err) + } + } + for index, received := range replayed { + message := waitKafkaMessage(t, received, 30*time.Second) + if string(message.Payload) != string(secondPayloads[index]) { + t.Fatalf("message[%d] replayed uncommitted payload %q, want %q", index, message.Payload, secondPayloads[index]) + } + } +} + +func waitKafkaMessage(t *testing.T, messages <-chan Message, timeout time.Duration) Message { + t.Helper() + select { + case message := <-messages: + return message + case <-time.After(timeout): + t.Fatal("timed out waiting for kafka message") + return Message{} + } +} diff --git a/pkg/mq/kafka_test.go b/pkg/mq/kafka_test.go new file mode 100644 index 0000000..bab1020 --- /dev/null +++ b/pkg/mq/kafka_test.go @@ -0,0 +1,234 @@ +package mq + +import ( + "context" + "errors" + "net" + "testing" + "time" + + kafkago "github.com/segmentio/kafka-go" + metadataapi "github.com/segmentio/kafka-go/protocol/metadata" +) + +type kafkaRoundTripperFunc func(context.Context, net.Addr, kafkago.Request) (kafkago.Response, error) + +func (f kafkaRoundTripperFunc) RoundTrip(ctx context.Context, addr net.Addr, request kafkago.Request) (kafkago.Response, error) { + return f(ctx, addr, request) +} + +func TestDisabledKafkaIsSafeAndUnavailable(t *testing.T) { + client, err := NewKafka(KafkaConfig{}) + if err != nil { + t.Fatal(err) + } + if client.Connected() { + t.Fatal("disabled kafka reported connected") + } + if err = client.Publish(context.Background(), "orders", []byte("test"), AtLeastOnce, false); !errors.Is(err, ErrUnavailable) { + t.Fatalf("publish error = %v", err) + } + if err = client.Close(); err != nil { + t.Fatal(err) + } +} + +func TestEnabledKafkaRequiresConnectionSettings(t *testing.T) { + if _, err := NewKafka(KafkaConfig{Enabled: true}); err == nil { + t.Fatal("enabled kafka without brokers should fail") + } + if _, err := NewKafka(KafkaConfig{Enabled: true, Brokers: []string{"localhost:9092"}}); err == nil { + t.Fatal("enabled kafka without client and group ids should fail") + } +} + +func TestKafkaConfigRejectsInvalidBrokerAndTLSSettings(t *testing.T) { + config := KafkaConfig{Enabled: true, Brokers: []string{"missing-port"}, ClientID: "kra", GroupID: "kra"} + if _, err := NewKafka(config); err == nil { + t.Fatal("invalid kafka broker was accepted") + } + config.Brokers = []string{"localhost:9092"} + config.TLSSkipVerify = true + if _, err := NewKafka(config); err == nil { + t.Fatal("tls skip verify without tls was accepted") + } +} + +func TestKafkaRejectsUnsupportedMessageSemantics(t *testing.T) { + client, err := NewKafka(KafkaConfig{}) + if err != nil { + t.Fatal(err) + } + if err = client.Publish(context.Background(), "orders", nil, AtMostOnce, true); err == nil { + t.Fatal("retained kafka message was accepted") + } + if err = client.Publish(context.Background(), "orders", nil, ExactlyOnce, false); err == nil { + t.Fatal("qos 2 kafka message was accepted") + } + if err = client.Subscribe(context.Background(), "orders", ExactlyOnce, func(context.Context, Message) {}); err == nil { + t.Fatal("qos 2 kafka subscription was accepted") + } +} + +func TestKafkaTopicNotReadyClassification(t *testing.T) { + for _, err := range []error{ + kafkago.UnknownTopicOrPartition, + kafkago.LeaderNotAvailable, + kafkago.NotLeaderForPartition, + kafkago.WriteErrors{kafkago.UnknownTopicOrPartition}, + } { + if !kafkaTopicNotReady(err) { + t.Fatalf("error %v was not classified as topic-not-ready", err) + } + } + for _, err := range []error{ + kafkago.TopicAuthorizationFailed, + errors.New("network failed"), + kafkago.WriteErrors{kafkago.UnknownTopicOrPartition, kafkago.TopicAuthorizationFailed}, + } { + if kafkaTopicNotReady(err) { + t.Fatalf("error %v was classified as topic-not-ready", err) + } + } +} + +func TestKafkaStartOffset(t *testing.T) { + if kafkaStartOffset("latest") != -1 { + t.Fatal("latest offset was not mapped to kafka last offset") + } + if kafkaStartOffset("earliest") != -2 { + t.Fatal("earliest offset was not mapped to kafka first offset") + } +} + +func TestNilKafkaClientIsSafe(t *testing.T) { + var client *Kafka + if err := client.Publish(context.Background(), "orders", nil, AtMostOnce, false); !errors.Is(err, ErrUnavailable) { + t.Fatalf("publish error = %v", err) + } + if err := client.Subscribe(context.Background(), "orders", AtMostOnce, func(context.Context, Message) {}); !errors.Is(err, ErrUnavailable) { + t.Fatalf("subscribe error = %v", err) + } + if err := client.Unsubscribe(context.Background(), "orders"); !errors.Is(err, ErrUnavailable) { + t.Fatalf("unsubscribe error = %v", err) + } + if err := client.Close(); err != nil { + t.Fatalf("close error = %v", err) + } +} + +func TestKafkaReconnectDelayExpires(t *testing.T) { + client := &Kafka{config: KafkaConfig{ReconnectInterval: 20 * time.Millisecond}} + client.markUnavailable() + if client.Connected() || !client.Reconnecting() { + t.Fatalf("failure state connected=%t reconnecting=%t", client.Connected(), client.Reconnecting()) + } + client.reconnectAt.Store(time.Now().Add(-time.Second).UnixNano()) + if client.Reconnecting() { + t.Fatal("reconnect delay did not expire") + } + client.markConnected() + if !client.Connected() || client.Reconnecting() { + t.Fatalf("recovered state connected=%t reconnecting=%t", client.Connected(), client.Reconnecting()) + } +} + +func TestKafkaUnsubscribeDoesNotWaitForConsumerHandler(t *testing.T) { + consumeCtx, cancel := context.WithCancel(context.Background()) + reader := kafkago.NewReader(kafkago.ReaderConfig{Brokers: []string{"127.0.0.1:9092"}, Topic: "orders", Partition: 0}) + client := &Kafka{subscriptions: map[string]*kafkaSubscription{ + "orders": {reader: reader, cancel: cancel}, + }} + done := make(chan error, 1) + go func() { done <- client.Unsubscribe(context.Background(), "orders") }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("unsubscribe waited for the consumer handler") + } + if consumeCtx.Err() == nil { + t.Fatal("consumer context was not canceled") + } +} + +func TestKafkaUnsubscribeRejectsBlankTopics(t *testing.T) { + client := &Kafka{subscriptions: make(map[string]*kafkaSubscription)} + if err := client.Unsubscribe(context.Background(), " "); err == nil { + t.Fatal("blank kafka topic was accepted") + } +} + +func TestProbeKafkaMetadataDoesNotAutoCreateTopic(t *testing.T) { + var captured *metadataapi.Request + client := &kafkago.Client{ + Addr: kafkago.TCP("kafka-1:9092", "kafka-2:9092"), + Transport: kafkaRoundTripperFunc(func(_ context.Context, _ net.Addr, request kafkago.Request) (kafkago.Response, error) { + captured = request.(*metadataapi.Request) + return &metadataapi.Response{ + Brokers: []metadataapi.ResponseBroker{{NodeID: 1, Host: "kafka-1", Port: 9092}}, + Topics: []metadataapi.ResponseTopic{{ + Name: "orders", + Partitions: []metadataapi.ResponsePartition{{ + PartitionIndex: 0, + LeaderID: 1, + }}, + }}, + }, nil + }), + } + if err := probeKafkaMetadata(context.Background(), client, "orders"); err != nil { + t.Fatal(err) + } + if captured == nil || len(captured.TopicNames) != 1 || captured.TopicNames[0] != "orders" { + t.Fatalf("metadata request = %#v", captured) + } + if captured.AllowAutoTopicCreation { + t.Fatal("subscription metadata lookup enabled automatic topic creation") + } +} + +func TestProbeKafkaMetadataRequestsBrokersWithoutListingTopics(t *testing.T) { + var captured *metadataapi.Request + client := &kafkago.Client{ + Addr: kafkago.TCP("kafka-1:9092"), + Transport: kafkaRoundTripperFunc(func(_ context.Context, _ net.Addr, request kafkago.Request) (kafkago.Response, error) { + captured = request.(*metadataapi.Request) + return &metadataapi.Response{Brokers: []metadataapi.ResponseBroker{{NodeID: 1, Host: "kafka-1", Port: 9092}}}, nil + }), + } + if err := probeKafkaMetadata(context.Background(), client, ""); err != nil { + t.Fatal(err) + } + if captured == nil || captured.TopicNames == nil || len(captured.TopicNames) != 0 { + t.Fatalf("broker metadata request topics = %#v", captured) + } +} + +func TestKafkaTopicMetadataErrorDoesNotMarkClusterUnavailable(t *testing.T) { + metadata := &kafkago.Client{ + Addr: kafkago.TCP("kafka-1:9092"), + Transport: kafkaRoundTripperFunc(func(_ context.Context, _ net.Addr, _ kafkago.Request) (kafkago.Response, error) { + return &metadataapi.Response{ + Brokers: []metadataapi.ResponseBroker{{NodeID: 1, Host: "kafka-1", Port: 9092}}, + Topics: []metadataapi.ResponseTopic{{Name: "missing", ErrorCode: int16(kafkago.UnknownTopicOrPartition)}}, + }, nil + }), + } + client := &Kafka{ + config: defaultKafkaConfig(KafkaConfig{}), + dialer: &kafkago.Dialer{}, + metadata: metadata, + subscriptions: make(map[string]*kafkaSubscription), + } + client.markUnavailable() + err := client.Subscribe(context.Background(), "missing", AtLeastOnce, func(context.Context, Message) {}) + if err == nil { + t.Fatal("missing kafka topic was accepted") + } + if !client.Connected() || client.Reconnecting() { + t.Fatalf("topic error changed cluster state: connected=%t reconnecting=%t", client.Connected(), client.Reconnecting()) + } +} diff --git a/pkg/mq/subscription.go b/pkg/mq/subscription.go index 350af86..ee44ab4 100644 --- a/pkg/mq/subscription.go +++ b/pkg/mq/subscription.go @@ -27,6 +27,9 @@ func NormalizeSubscriptionSet(set SubscriptionSet) (SubscriptionSet, error) { if set.Topics[index].QoS > ExactlyOnce { return SubscriptionSet{}, fmt.Errorf("invalid mq qos %d for topic %q", set.Topics[index].QoS, set.Topics[index].Topic) } + if set.Provider != ProviderEMQX && set.Topics[index].QoS > AtLeastOnce { + return SubscriptionSet{}, fmt.Errorf("mq provider %q supports qos 0 or 1 for topic %q", set.Provider, set.Topics[index].Topic) + } if set.Topics[index].Handler == nil { return SubscriptionSet{}, fmt.Errorf("mq subscription handler is nil for topic %q", set.Topics[index].Topic) } diff --git a/pkg/mq/subscription_test.go b/pkg/mq/subscription_test.go index eb45f93..fbebe41 100644 --- a/pkg/mq/subscription_test.go +++ b/pkg/mq/subscription_test.go @@ -16,7 +16,7 @@ func TestNormalizeSubscriptionSetValidDeclaration(t *testing.T) { Topics: []TopicSubscription{ {Topic: " orders.created ", QoS: AtMostOnce, Handler: handler}, {Topic: "orders.updated", QoS: AtLeastOnce, Handler: handler}, - {Topic: "orders.deleted", QoS: ExactlyOnce, Handler: handler}, + {Topic: "orders.deleted", QoS: AtLeastOnce, Handler: handler}, }, } @@ -31,7 +31,7 @@ func TestNormalizeSubscriptionSetValidDeclaration(t *testing.T) { Topics: []TopicSubscription{ {Topic: "orders.created", QoS: AtMostOnce, Handler: handler}, {Topic: "orders.updated", QoS: AtLeastOnce, Handler: handler}, - {Topic: "orders.deleted", QoS: ExactlyOnce, Handler: handler}, + {Topic: "orders.deleted", QoS: AtLeastOnce, Handler: handler}, }, } if got.Owner != want.Owner || got.Provider != want.Provider { @@ -53,6 +53,34 @@ func TestNormalizeSubscriptionSetValidDeclaration(t *testing.T) { } } +func TestNormalizeSubscriptionSetAcceptsMQTTQoS2(t *testing.T) { + set, err := NormalizeSubscriptionSet(SubscriptionSet{ + Owner: "devices", + Provider: ProviderEMQX, + Topics: []TopicSubscription{{Topic: "devices/status", QoS: ExactlyOnce, Handler: func(context.Context, Message) {}}}, + }) + if err != nil { + t.Fatal(err) + } + if set.Topics[0].QoS != ExactlyOnce { + t.Fatalf("qos = %d, want %d", set.Topics[0].QoS, ExactlyOnce) + } +} + +func TestNormalizeSubscriptionSetAcceptsKafka(t *testing.T) { + set, err := NormalizeSubscriptionSet(SubscriptionSet{ + Owner: "orders", + Provider: " KAFKA ", + Topics: []TopicSubscription{{Topic: "orders.created", QoS: AtLeastOnce, Handler: func(context.Context, Message) {}}}, + }) + if err != nil { + t.Fatal(err) + } + if set.Provider != ProviderKafka { + t.Fatalf("provider = %q, want %q", set.Provider, ProviderKafka) + } +} + func TestNormalizeSubscriptionSetRejectsInvalidDeclarations(t *testing.T) { handler := func(context.Context, Message) {} base := func() SubscriptionSet { @@ -82,7 +110,7 @@ func TestNormalizeSubscriptionSetRejectsInvalidDeclarations(t *testing.T) { }, { name: "unsupported provider", - set: func() SubscriptionSet { set := base(); set.Provider = "kafka"; return set }(), + set: func() SubscriptionSet { set := base(); set.Provider = "pulsar"; return set }(), wantErr: "unsupported mq provider", }, { @@ -128,6 +156,26 @@ func TestNormalizeSubscriptionSetRejectsInvalidDeclarations(t *testing.T) { }(), wantErr: "invalid mq qos", }, + { + name: "kafka qos 2", + set: func() SubscriptionSet { + set := base() + set.Provider = ProviderKafka + set.Topics[0].QoS = ExactlyOnce + return set + }(), + wantErr: "supports qos 0 or 1", + }, + { + name: "rabbitmq qos 2", + set: func() SubscriptionSet { + set := base() + set.Provider = ProviderRabbitMQ + set.Topics[0].QoS = ExactlyOnce + return set + }(), + wantErr: "supports qos 0 or 1", + }, } for _, test := range tests { diff --git a/web/src/view/systemTools/integration/config.vue b/web/src/view/systemTools/integration/config.vue index c1aa3e9..3a75ffa 100644 --- a/web/src/view/systemTools/integration/config.vue +++ b/web/src/view/systemTools/integration/config.vue @@ -299,7 +299,7 @@ const NUMBER_CONSTRAINTS = { heartbeat: { min: 0 }, min_bytes: { min: 1, integer: true }, max_bytes: { min: 1, integer: true }, - max_wait: { min: 1 }, + max_wait: { min: 1, integer: true }, max_message_size: { min: 0 }, message_buffer_size: { min: 0 } } @@ -569,7 +569,6 @@ const persist = async (item, operation) => { config: item.config }) if (res.code !== 0) { - ElMessage.error(res.msg || '保存失败') return false } markSaved(item) @@ -600,12 +599,11 @@ const testSelected = async () => { config: item.config }) if (res.code !== 0) { - ElMessage.error(res.msg || '连接测试失败') return } ElMessage.success(`${item.name || providerMeta(item).name} 连接测试成功`) } catch { - ElMessage.error('连接测试失败') + // The request layer owns transport and server error presentation. } finally { delete pending[key] }