Copilot commented on code in PR #862:
URL: https://github.com/apache/dubbo-go-pixiu/pull/862#discussion_r3084090331
##########
go.mod:
##########
@@ -4,7 +4,7 @@ go 1.25
require (
dario.cat/mergo v1.0.1
- dubbo.apache.org/dubbo-go/v3 v3.3.1
+ dubbo.apache.org/dubbo-go/v3 v3.0.0-20260210015753-35ea886421f9
github.com/IBM/sarama v1.45.1
Review Comment:
PR description says the client is updated to "latest dubbo-go v3.3.1", but
`go.mod` pins `dubbo.apache.org/dubbo-go/v3` to the pseudo-version
`v3.0.0-20260210015753-35ea886421f9` instead. Please align the dependency
version with the PR description (or update the description and rationale if
this pseudo-version is required).
##########
pkg/client/dubbo/dubbo.go:
##########
@@ -289,79 +354,342 @@ func (dc *Client) check(key string) bool {
return false
}
-// Get find a dubbo GenericService
-func (dc *Client) Get(ir config.IntegrationRequest) *generic.GenericService {
- key := apiKey(&ir)
- if dc.check(key) {
- return dc.get(key)
+func (dc *Client) resolveReferSpec(irequest config.IntegrationRequest)
(resolvedReferSpec, error) {
+ spec := resolvedReferSpec{
+ Interface: irequest.Interface,
+ Group: irequest.Group,
+ Version: irequest.Version,
+ ConsumerDefaults: dc.resolveConsumerDefaults(irequest),
}
- return dc.create(key, ir)
-}
+ if len(dc.registries) > 0 {
+ registryIDs := make([]string, 0, len(dc.registries))
+ useNacosWarmup := false
+ for id, registry := range dc.registries {
+ registryIDs = append(registryIDs, id)
+ if registry != nil && registry.Protocol == "nacos" {
+ useNacosWarmup = true
+ }
+ }
+ sort.Strings(registryIDs)
+ spec.Mode = "registry"
+ spec.RegistryIDs = registryIDs
+ spec.UseNacosWarmup = useNacosWarmup
+ spec.EffectiveProtocol = resolveDeclaredProtocol(irequest)
+ return spec, nil
+ }
-func apiKey(ir *config.IntegrationRequest) string {
- dbc := ir.DubboBackendConfig
- return strings.Join([]string{dbc.ClusterName, dbc.ApplicationName,
dbc.Interface, dbc.Version, dbc.Group}, "_")
+ canonicalURL := strings.TrimSpace(irequest.URL)
+ if canonicalURL == "" {
+ return resolvedReferSpec{}, errors.New("dubbo refer mode
invalid: no registry configured and no direct url provided")
+ }
+ directProtocol, err := directURLProtocol(canonicalURL)
+ if err != nil {
+ return resolvedReferSpec{}, err
+ }
+
+ spec.Mode = "direct"
+ spec.URL = canonicalURL
+ spec.EffectiveProtocol = directProtocol
+ return spec, nil
}
-func (dc *Client) create(key string, irequest config.IntegrationRequest)
*generic.GenericService {
- useNacosRegister := false
- registerIds := make([]string, 0)
- for k, v := range dc.rootConfig.Registries {
- registerIds = append(registerIds, k)
- if v.Protocol == "nacos" {
- useNacosRegister = true
+func (dc *Client) resolveConsumerDefaults(irequest config.IntegrationRequest)
resolvedConsumerDefaults {
+ defaults := resolvedConsumerDefaults{
+ Cluster: "failover",
+ Retries: "3",
+ }
+
+ if dc.dubboProxyConfig == nil {
+ if strings.TrimSpace(irequest.Retries) != "" {
+ defaults.Retries = strings.TrimSpace(irequest.Retries)
}
+ defaults.RequestTimeout = cst.DefaultReqTimeout
+ return defaults
}
- refConf := dg.ReferenceConfig{
- InterfaceName: irequest.Interface,
- Cluster: constant.ClusterKeyFailover,
- RegistryIDs: registerIds,
- Protocol: dubbo.DUBBO,
- Generic: "true",
- Version: irequest.Version,
- Group: irequest.Group,
- Loadbalance: dc.dubboProxyConfig.LoadBalance,
- Retries: dc.dubboProxyConfig.Retries,
+ defaults.LoadBalance = dc.dubboProxyConfig.LoadBalance
+ if strings.TrimSpace(dc.dubboProxyConfig.Retries) != "" {
+ defaults.Retries =
strings.TrimSpace(dc.dubboProxyConfig.Retries)
+ } else if strings.TrimSpace(irequest.Retries) != "" {
+ defaults.Retries = strings.TrimSpace(irequest.Retries)
}
- if refConf.Retries == "" {
- if len(irequest.Retries) == 0 {
- refConf.Retries = "3"
+ if dc.dubboProxyConfig.Timeout != nil {
+ if timeout, err :=
time.ParseDuration(dc.dubboProxyConfig.Timeout.RequestTimeoutStr); err == nil {
+ defaults.RequestTimeout = timeout
} else {
- refConf.Retries = irequest.Retries
+ defaults.RequestTimeout = cst.DefaultReqTimeout
}
+ } else {
+ defaults.RequestTimeout = cst.DefaultReqTimeout
}
- if dc.dubboProxyConfig.Timeout != nil {
- refConf.RequestTimeout =
dc.dubboProxyConfig.Timeout.RequestTimeoutStr
- } else {
- refConf.RequestTimeout = cst.DefaultReqTimeout.String()
+ return defaults
+}
+
+func (spec resolvedReferSpec) validate() error {
+ switch spec.Mode {
+ case "registry":
+ if len(spec.RegistryIDs) == 0 {
+ return errors.New("dubbo refer mode invalid: registry
mode requires registry ids")
+ }
+ return nil
+ case "direct":
+ if strings.TrimSpace(spec.URL) == "" {
+ return errors.New("dubbo refer mode invalid: direct
mode requires direct url")
+ }
+ return nil
+ default:
+ return errors.Errorf("dubbo refer mode invalid: %s", spec.Mode)
+ }
+}
+
+func (spec resolvedReferSpec) cacheKey() (string, error) {
+ if err := spec.validate(); err != nil {
+ return "", err
+ }
+
+ key := spec.genericServiceKey()
+ raw, err := json.Marshal(key)
+ if err != nil {
+ return "", errors.Wrap(err, "marshal generic service key")
+ }
+ return string(raw), nil
+}
+
+func (spec resolvedReferSpec) genericServiceKey() genericServiceKey {
+ registryIDs := append([]string(nil), spec.RegistryIDs...)
+ sort.Strings(registryIDs)
+ return genericServiceKey{
+ Mode: spec.Mode,
+ URL: spec.URL,
+ RegistryIDs: registryIDs,
+ Cluster: spec.ConsumerDefaults.Cluster,
+ Interface: spec.Interface,
+ Version: spec.Version,
+ Group: spec.Group,
+ EffectiveProtocol: spec.EffectiveProtocol,
+ }
+}
+
+// Get find a dubbo GenericService
+func (dc *Client) Get(spec resolvedReferSpec) (*generic.GenericService, error)
{
+ key, err := spec.cacheKey()
+ if err != nil {
+ return nil, err
+ }
+ if dc.check(key) {
+ return dc.get(key), nil
+ }
+
+ return dc.create(spec)
+}
+
+func (dc *Client) create(spec resolvedReferSpec) (*generic.GenericService,
error) {
+ if err := spec.validate(); err != nil {
+ return nil, err
+ }
+ if dc.dubboClient == nil {
+ return nil, errors.New("dubbo client is not initialized, call
Apply() first")
+ }
+
+ key, err := spec.cacheKey()
+ if err != nil {
+ return nil, err
+ }
+
+ opts, err := dc.buildReferenceOptions(spec)
+ if err != nil {
+ return nil, err
}
- logger.Debugf("[dubbo-go-pixiu] client dubbo timeout val %v",
refConf.RequestTimeout)
+
dc.lock.Lock()
defer dc.lock.Unlock()
if service, ok := dc.GenericServicePool[key]; ok {
- return service
+ return service, nil
}
- if err := dg.Load(dg.WithRootConfig(dc.rootConfig)); err != nil {
- panic(err)
+ clientService, err := dc.dubboClient.NewGenericService(spec.Interface,
opts...)
+ if err != nil {
+ return nil, err
}
- _ = refConf.Init(dc.rootConfig)
- refConf.GenericLoad(key)
-
- // sleep when first call to fetch enough service meta data from nacos
- // todo: GenericLoad should guarantee it
- if useNacosRegister {
- time.Sleep(1000 * time.Millisecond)
+ if spec.Mode == "registry" && spec.UseNacosWarmup {
+ time.Sleep(time.Second)
}
Review Comment:
`create` holds `dc.lock` while calling
`dc.dubboClient.NewGenericService(...)` and also while doing the Nacos warmup
`time.Sleep(time.Second)`. This blocks all concurrent lookups/creations in
`GenericServicePool` for up to 1s and can become a latency/throughput
bottleneck. Consider narrowing the critical section (double-check cache under
lock, create outside the lock, then lock again to publish) and ensure any
warmup sleep happens outside the mutex.
##########
pkg/client/dubbo/dubbo.go:
##########
@@ -289,79 +354,342 @@ func (dc *Client) check(key string) bool {
return false
}
-// Get find a dubbo GenericService
-func (dc *Client) Get(ir config.IntegrationRequest) *generic.GenericService {
- key := apiKey(&ir)
- if dc.check(key) {
- return dc.get(key)
+func (dc *Client) resolveReferSpec(irequest config.IntegrationRequest)
(resolvedReferSpec, error) {
+ spec := resolvedReferSpec{
+ Interface: irequest.Interface,
+ Group: irequest.Group,
+ Version: irequest.Version,
+ ConsumerDefaults: dc.resolveConsumerDefaults(irequest),
}
- return dc.create(key, ir)
-}
+ if len(dc.registries) > 0 {
+ registryIDs := make([]string, 0, len(dc.registries))
+ useNacosWarmup := false
+ for id, registry := range dc.registries {
+ registryIDs = append(registryIDs, id)
+ if registry != nil && registry.Protocol == "nacos" {
+ useNacosWarmup = true
+ }
+ }
+ sort.Strings(registryIDs)
+ spec.Mode = "registry"
+ spec.RegistryIDs = registryIDs
+ spec.UseNacosWarmup = useNacosWarmup
+ spec.EffectiveProtocol = resolveDeclaredProtocol(irequest)
+ return spec, nil
+ }
-func apiKey(ir *config.IntegrationRequest) string {
- dbc := ir.DubboBackendConfig
- return strings.Join([]string{dbc.ClusterName, dbc.ApplicationName,
dbc.Interface, dbc.Version, dbc.Group}, "_")
+ canonicalURL := strings.TrimSpace(irequest.URL)
+ if canonicalURL == "" {
+ return resolvedReferSpec{}, errors.New("dubbo refer mode
invalid: no registry configured and no direct url provided")
+ }
+ directProtocol, err := directURLProtocol(canonicalURL)
+ if err != nil {
+ return resolvedReferSpec{}, err
+ }
+
+ spec.Mode = "direct"
+ spec.URL = canonicalURL
+ spec.EffectiveProtocol = directProtocol
+ return spec, nil
}
-func (dc *Client) create(key string, irequest config.IntegrationRequest)
*generic.GenericService {
- useNacosRegister := false
- registerIds := make([]string, 0)
- for k, v := range dc.rootConfig.Registries {
- registerIds = append(registerIds, k)
- if v.Protocol == "nacos" {
- useNacosRegister = true
+func (dc *Client) resolveConsumerDefaults(irequest config.IntegrationRequest)
resolvedConsumerDefaults {
+ defaults := resolvedConsumerDefaults{
+ Cluster: "failover",
+ Retries: "3",
+ }
+
+ if dc.dubboProxyConfig == nil {
+ if strings.TrimSpace(irequest.Retries) != "" {
+ defaults.Retries = strings.TrimSpace(irequest.Retries)
}
+ defaults.RequestTimeout = cst.DefaultReqTimeout
+ return defaults
}
- refConf := dg.ReferenceConfig{
- InterfaceName: irequest.Interface,
- Cluster: constant.ClusterKeyFailover,
- RegistryIDs: registerIds,
- Protocol: dubbo.DUBBO,
- Generic: "true",
- Version: irequest.Version,
- Group: irequest.Group,
- Loadbalance: dc.dubboProxyConfig.LoadBalance,
- Retries: dc.dubboProxyConfig.Retries,
+ defaults.LoadBalance = dc.dubboProxyConfig.LoadBalance
+ if strings.TrimSpace(dc.dubboProxyConfig.Retries) != "" {
+ defaults.Retries =
strings.TrimSpace(dc.dubboProxyConfig.Retries)
+ } else if strings.TrimSpace(irequest.Retries) != "" {
+ defaults.Retries = strings.TrimSpace(irequest.Retries)
}
- if refConf.Retries == "" {
- if len(irequest.Retries) == 0 {
- refConf.Retries = "3"
+ if dc.dubboProxyConfig.Timeout != nil {
+ if timeout, err :=
time.ParseDuration(dc.dubboProxyConfig.Timeout.RequestTimeoutStr); err == nil {
+ defaults.RequestTimeout = timeout
} else {
- refConf.Retries = irequest.Retries
+ defaults.RequestTimeout = cst.DefaultReqTimeout
}
+ } else {
+ defaults.RequestTimeout = cst.DefaultReqTimeout
}
- if dc.dubboProxyConfig.Timeout != nil {
- refConf.RequestTimeout =
dc.dubboProxyConfig.Timeout.RequestTimeoutStr
- } else {
- refConf.RequestTimeout = cst.DefaultReqTimeout.String()
+ return defaults
+}
+
+func (spec resolvedReferSpec) validate() error {
+ switch spec.Mode {
+ case "registry":
+ if len(spec.RegistryIDs) == 0 {
+ return errors.New("dubbo refer mode invalid: registry
mode requires registry ids")
+ }
+ return nil
+ case "direct":
+ if strings.TrimSpace(spec.URL) == "" {
+ return errors.New("dubbo refer mode invalid: direct
mode requires direct url")
+ }
+ return nil
+ default:
+ return errors.Errorf("dubbo refer mode invalid: %s", spec.Mode)
+ }
+}
+
+func (spec resolvedReferSpec) cacheKey() (string, error) {
+ if err := spec.validate(); err != nil {
+ return "", err
+ }
+
+ key := spec.genericServiceKey()
+ raw, err := json.Marshal(key)
+ if err != nil {
+ return "", errors.Wrap(err, "marshal generic service key")
+ }
+ return string(raw), nil
+}
+
+func (spec resolvedReferSpec) genericServiceKey() genericServiceKey {
+ registryIDs := append([]string(nil), spec.RegistryIDs...)
+ sort.Strings(registryIDs)
+ return genericServiceKey{
+ Mode: spec.Mode,
+ URL: spec.URL,
+ RegistryIDs: registryIDs,
+ Cluster: spec.ConsumerDefaults.Cluster,
+ Interface: spec.Interface,
+ Version: spec.Version,
+ Group: spec.Group,
+ EffectiveProtocol: spec.EffectiveProtocol,
+ }
+}
+
+// Get find a dubbo GenericService
+func (dc *Client) Get(spec resolvedReferSpec) (*generic.GenericService, error)
{
+ key, err := spec.cacheKey()
+ if err != nil {
+ return nil, err
+ }
+ if dc.check(key) {
+ return dc.get(key), nil
+ }
+
+ return dc.create(spec)
+}
+
+func (dc *Client) create(spec resolvedReferSpec) (*generic.GenericService,
error) {
+ if err := spec.validate(); err != nil {
+ return nil, err
+ }
+ if dc.dubboClient == nil {
+ return nil, errors.New("dubbo client is not initialized, call
Apply() first")
+ }
+
+ key, err := spec.cacheKey()
+ if err != nil {
+ return nil, err
+ }
+
+ opts, err := dc.buildReferenceOptions(spec)
+ if err != nil {
+ return nil, err
}
- logger.Debugf("[dubbo-go-pixiu] client dubbo timeout val %v",
refConf.RequestTimeout)
+
dc.lock.Lock()
defer dc.lock.Unlock()
if service, ok := dc.GenericServicePool[key]; ok {
- return service
+ return service, nil
}
- if err := dg.Load(dg.WithRootConfig(dc.rootConfig)); err != nil {
- panic(err)
+ clientService, err := dc.dubboClient.NewGenericService(spec.Interface,
opts...)
+ if err != nil {
+ return nil, err
}
- _ = refConf.Init(dc.rootConfig)
- refConf.GenericLoad(key)
-
- // sleep when first call to fetch enough service meta data from nacos
- // todo: GenericLoad should guarantee it
- if useNacosRegister {
- time.Sleep(1000 * time.Millisecond)
+ if spec.Mode == "registry" && spec.UseNacosWarmup {
+ time.Sleep(time.Second)
}
- clientService := refConf.GetRPCService().(*generic.GenericService)
dc.GenericServicePool[key] = clientService
- return clientService
+ return clientService, nil
+}
+
+// buildReferenceOptions builds a list of dubbo-go ReferenceOption using the
official API.
+func (dc *Client) buildReferenceOptions(spec resolvedReferSpec)
([]dclient.ReferenceOption, error) {
+ if err := spec.validate(); err != nil {
+ return nil, err
+ }
+ if strings.TrimSpace(spec.EffectiveProtocol) == "" {
+ return nil, errors.New("dubbo refer mode invalid: effective
protocol is required")
+ }
+
+ defaults := spec.ConsumerDefaults
+ opts := make([]dclient.ReferenceOption, 0, 16)
+
+ opts = append(opts, dclient.WithInterface(spec.Interface))
+ if spec.Group != "" {
+ opts = append(opts, dclient.WithGroup(spec.Group))
+ }
+ if spec.Version != "" {
+ opts = append(opts, dclient.WithVersion(spec.Version))
+ }
+
+ opts = appendModeReferenceOptions(opts, spec)
+ opts = append(opts, clusterReferenceOption(defaults.Cluster))
+ opts = append(opts, protocolReferenceOption(spec.EffectiveProtocol))
+
+ if loadBalanceOpt := loadBalanceReferenceOption(defaults.LoadBalance);
loadBalanceOpt != nil {
+ opts = append(opts, loadBalanceOpt)
+ }
+
+ retries := 3
+ if strings.TrimSpace(defaults.Retries) != "" {
+ if resolvedRetries, err := strconv.Atoi(defaults.Retries); err
== nil {
+ retries = resolvedRetries
+ }
+ }
+ opts = append(opts, dclient.WithRetries(retries))
+
+ timeout := defaults.RequestTimeout
+ if timeout <= 0 {
+ timeout = cst.DefaultReqTimeout
+ }
+ opts = append(opts, dclient.WithRequestTimeout(timeout))
+
+ opts = append(opts, dclient.WithGeneric())
+
+ return opts, nil
+}
+
+func appendModeReferenceOptions(opts []dclient.ReferenceOption, spec
resolvedReferSpec) []dclient.ReferenceOption {
+ switch spec.Mode {
+ case "registry":
+ registryIDs := append([]string(nil), spec.RegistryIDs...)
+ sort.Strings(registryIDs)
+ return append(opts, dclient.WithRegistryIDs(registryIDs...))
+ case "direct":
+ return append(opts, dclient.WithURL(spec.URL))
+ default:
+ return opts
+ }
+}
+
+func clusterReferenceOption(cluster string) dclient.ReferenceOption {
+ switch cluster {
+ case "failover":
+ return dclient.WithClusterFailOver()
+ case "failfast":
+ return dclient.WithClusterFailFast()
+ case "failsafe":
+ return dclient.WithClusterFailSafe()
+ case "failback":
+ return dclient.WithClusterFailBack()
+ case "broadcast":
+ return dclient.WithClusterBroadcast()
+ case "forking":
+ return dclient.WithClusterForking()
+ case "available":
+ return dclient.WithClusterAvailable()
+ case "zoneaware":
+ return dclient.WithClusterZoneAware()
+ case "adaptiveservice":
+ return dclient.WithClusterAdaptiveService()
+ default:
+ return dclient.WithCluster(cluster)
+ }
+}
+
+func protocolReferenceOption(protocol string) dclient.ReferenceOption {
+ switch protocol {
+ case "tri", "triple":
+ return dclient.WithProtocolTriple()
+ case "dubbo":
+ return dclient.WithProtocolDubbo()
+ case "jsonrpc":
+ return dclient.WithProtocolJsonRPC()
+ default:
+ return dclient.WithProtocol(protocol)
+ }
+}
+
+func loadBalanceReferenceOption(loadBalance string) dclient.ReferenceOption {
+ switch loadBalance {
+ case "":
+ return nil
+ case "random":
+ return dclient.WithLoadBalanceRandom()
+ case "roundrobin":
+ return dclient.WithLoadBalanceRoundRobin()
+ case "leastactive":
+ return dclient.WithLoadBalanceLeastActive()
+ case "consistenthash", "consistenthashing":
+ return dclient.WithLoadBalanceConsistentHashing()
+ case "p2c":
+ return dclient.WithLoadBalanceP2C()
+ default:
+ return dclient.WithLoadBalance(loadBalance)
+ }
+}
+
+func resolveDeclaredProtocol(irequest config.IntegrationRequest) string {
+ if protocol := normalizeReferenceProtocol(irequest.Protocol); protocol
!= "" {
+ return protocol
+ }
+ if normalizeReferenceProtocol(irequest.RequestType) == "tri" {
+ return "tri"
+ }
+ return "dubbo"
+}
+
+func directURLProtocol(rawURL string) (string, error) {
+ parsedURL, err := url.Parse(rawURL)
+ if err != nil {
+ return "", errors.Wrapf(err, "parse direct url %q", rawURL)
+ }
+ return normalizeReferenceProtocol(parsedURL.Scheme), nil
+}
Review Comment:
`directURLProtocol` uses `url.Parse(rawURL)` and then reads
`parsedURL.Scheme`. For direct addresses like `127.0.0.1:20880` (missing
`://`), `url.Parse` treats `127.0.0.1` as the scheme, so this returns an
invalid "protocol" and later builds a reference with
`WithProtocol("127.0.0.1")`. Please detect missing schemes (e.g., no `://`) and
fall back to a sensible default (like the declared request protocol /
requestType, or "dubbo"), or validate the parsed scheme against known protocols
and error clearly.
##########
tools/benchmark/go.mod:
##########
@@ -3,7 +3,7 @@ module github.com/apache/dubbo-go-pixiu/tools/benchmark
go 1.25.0
require (
- dubbo.apache.org/dubbo-go/v3 v3.3.1
+ dubbo.apache.org/dubbo-go/v3 v3.0.0-20260210015753-35ea886421f9
github.com/apache/dubbo-go-hessian2 v1.12.5
Review Comment:
Same as the root module: this benchmark module is now pinned to
`dubbo.apache.org/dubbo-go/v3 v3.0.0-20260210015753-35ea886421f9`, which
conflicts with the PR description calling out v3.3.1. Please keep the benchmark
module aligned with the intended dubbo-go version (or update the PR
description/rationale accordingly).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
