This is an automated email from the ASF dual-hosted git repository.
AlexStocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new fc7f14f35 feat(metadata): Clarify metadata report selection for
multi-registry and multi-instance scenarios (#3369)
fc7f14f35 is described below
commit fc7f14f35369350e8d319fe9bbd6deaa4b0d259c
Author: Nene7ko_ <[email protected]>
AuthorDate: Wed Jun 10 12:58:29 2026 +0800
feat(metadata): Clarify metadata report selection for multi-registry and
multi-instance scenarios (#3369)
* fix(metadata): make GetMetadataReport() deterministic in multi-registry
setup
* fix(metadata): Remove() fans out to all reports, consistent with Map()
* fix(metadata): thread registryId through GetMetadataFromMetadataReport
call chain
* fix(metadata): scope revision calculation to per-registry services,
thread registryId through createInstance
* fix(metadata): GetMetadataReportByRegistry returns nil for unknown
registryId instead of silently falling back to wrong registry
When a caller provides a specific non-empty registryId that has no
registered report, the previous implementation silently fell back to
GetMetadataReport() and returned an arbitrary report from a different
registry. This undermined the per-registry scoping added in the earlier
commits: metadata for registry A could be fetched from registry B's
report without any indication.
Fix: return nil (with a Warnf log) when a specific registryId is not
found. The existing nil-guard in GetMetadataFromMetadataReport (client.go)
already surfaces this as an error to the caller, which is the correct
behaviour.
The empty-string path is unchanged: it still delegates to
GetMetadataReport() because callers that pass "" have no registry
context and should receive the stable default.
Also expose ClearMetadataReportInstances() to allow cross-package test
isolation of the package-level instances map.
* fix(metadata): ServiceNameMapping.Remove collects all errors instead of
returning only the last one
The previous implementation used a lastErr variable that was overwritten
on each iteration. If the first report failed and the second succeeded,
the error was silently discarded and the caller saw no error despite a
partial failure — leaving a stale mapping in the first registry.
Fix: collect all errors with errors.Join so the caller receives the full
failure picture. The loop continues past individual failures (best-effort
fan-out) so a transient error in one registry does not prevent removal
from the others.
Add a comment explaining the intentional behavioural difference from
Map(), which is fail-fast: Map stops at the first report failure, while
Remove continues and collects all errors so every reachable registry gets
the removal attempt.
* test: expand coverage for per-registry metadata report selection
Three gaps identified in PR review, all addressed:
1. metadata/client_test.go — rewrite TestGetMetadataFromMetadataReport
- Add sub-test: specific registryId routes to its own report (not
default)
- Add sub-test: unknown registryId returns error, not silent
wrong-registry result
- Each sub-test now resets instances independently to prevent state
leakage
2. registry/servicediscovery/customizer/service_revision_customizer_test.go
- Add TestExportedRevisionMissingRegistryIdYieldsZero: instance with no
RegistryIdKey gets revision "0" and does not borrow another registry's
service list
- Add TestSubscribedRevisionMissingRegistryIdYieldsZero: same for
subscribed
- Add t.Cleanup(ClearMetadataReportInstances) + unique registry-id
prefixes
to all four tests to prevent cross-test global map pollution
3. registry/servicediscovery/service_instances_changed_listener_impl_test.go
- Add TestListenerUsesRegistryIdToFetchRemoteMetadata: creates a listener
with registryId="remote-reg-test", registers a mock MetadataReport
under
that id, triggers OnEvent with a RemoteMetadataStorageType instance,
and
asserts via mock.AssertExpectations that the correct per-registry
report
was called — end-to-end proof that registryId threads from
NewServiceInstancesChangedListener through to
GetMetadataFromMetadataReport
* chore: translate Chinese comments to English, remove UTF-8 BOM
- metadata/report_instance_test.go: translate inline comments and
testify assertion messages in TestGetMetadataReportIsDeterministic
from Chinese to English, consistent with the rest of the file
- registry/servicediscovery/customizer/service_revision_customizer.go:
translate Customize doc-comments for both exported and subscribed
customizers from Chinese to English
- registry/servicediscovery/service_instances_changed_listener_impl_test.go:
strip the UTF-8 BOM (EF BB BF) that was introduced at line 1;
Go toolchain accepts BOMs but they are not idiomatic and cause
spurious diffs in some editors and CI tools
* style(metadata): fix import formatting
* fix(registry): upgrade missing-registryId log from Warn to Error with
accurate consequence
The previous Warnf said 'revision will be empty', which is wrong on two
counts: the actual computed revision is "0" (not empty), and revision
"0" causes OnEvent to silently skip the instance entirely, making it
permanently invisible to all consumers.
Upgrade to Errorf and describe the actual outcome so operators can
diagnose misconfigurations immediately. Applies to both exported and
subscribed revision customizers.
* fix(registry): scope metaCache key to (registryId, revision) to prevent
cross-registry metadata poisoning
The cache in GetMetadataInfo was keyed by revision string only. In a
multi-registry setup, if two registries happen to produce the same
revision value (e.g. identical service sets yielding the same CRC, or
test fixtures using literal revision strings), the second registry's
listener would get a cache hit and return MetadataInfo fetched via a
different registry's report — silently serving wrong endpoints with no
error surfaced.
Fix: use registryId+":"+revision as the cache key for both Get and Set,
so each registry's metadata is cached independently.
Update all test helpers and direct metaCache.Set/Delete calls to use
the same composite key (with constant.DefaultKey as the registryId for
all existing single-registry test listeners).
* docs(registry): restore TODO for multi-instance metadata service URL
alignment
The previous commit removed the TODO without fixing the underlying issue.
GetMetadataService() still returns a global singleton, so all instances
across registries receive the same metadata service URL regardless of
which registry they belong to.
Restore the TODO with a more detailed explanation so the gap is visible
to future contributors. The fix requires per-registry MetadataService
support and is out of scope for this PR.
* Revert "docs(registry): restore TODO for multi-instance metadata service
URL alignment"
This reverts commit 5d7bcc6cf69f6d05646c4e61beda666fc31ad4eb.
* style(metadata): fix import formatting
* fix(debug) : fix review problems
* style(metadata): fix format issue && revert some changes
* fix(metadata): update tests and comment to match default fallback in
GetMetadataReportByRegistry
GetMetadataReportByRegistry intentionally falls back to the 'default'
report when a specific registryId is not found. This supports standalone
metadata-report configs registered under 'default' being used by named
registries (e.g. nacos, zk). Update tests and doc comment to reflect
this behavior instead of the old 'return nil for unknown id' contract.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
* fix: address review comments on clarify-metadata-report-selection
- fix(registry): unify registryId default to constant.DefaultKey before
GetMetadataInfo lookup in RegisterService, eliminating empty-string
fallback mismatch and redundant GetParam calls
- fix(metadata): scope metaCache key to app:registryId:revision to
prevent cross-registry cache collisions
- fix(metadata/mapping): bind MappingListener to deterministic primary
report via GetMetadataReport() instead of non-deterministic i==0
- fix(metadata): include registryId in GetMetadataFromMetadataReport
error message for easier misconfiguration diagnosis
- test: add t.Cleanup to remove AddService/AddSubscribeURL entries and
prevent global state leakage between tests
- docs: restore TODO comment explaining GetMetadataService() singleton
limitation in multi-registry metadata service URL alignment
* fix: fix test isolation and cache key format after review changes
- fix(test): update metaCache keys in
service_instances_changed_listener_impl_test
to match new app:registryId:revision format
- test: use t.Name()+UnixNano as regID in service_discovery_registry_test
to prevent cross-test global state pollution without expanding production
API
---------
Co-authored-by: xnlemon <[email protected]>
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
metadata/client.go | 6 +-
metadata/client_test.go | 59 ++++++-
metadata/mapping/metadata/service_name_mapping.go | 50 +++++-
.../mapping/metadata/service_name_mapping_test.go | 100 ++++++++++++
metadata/report_instance.go | 42 ++++-
metadata/report_instance_test.go | 56 ++++++-
registry/nacos/service_discovery_test.go | 2 +-
.../metadata_service_url_params_customizer.go | 5 +-
.../customizer/service_revision_customizer.go | 35 +++--
.../customizer/service_revision_customizer_test.go | 173 +++++++++++++++++++++
.../servicediscovery/service_discovery_registry.go | 22 +--
.../service_discovery_registry_test.go | 16 +-
.../service_instances_changed_listener_impl.go | 18 ++-
...service_instances_changed_listener_impl_test.go | 129 +++++++++++++--
14 files changed, 642 insertions(+), 71 deletions(-)
diff --git a/metadata/client.go b/metadata/client.go
index 7d49e92ef..f6b1ffd33 100644
--- a/metadata/client.go
+++ b/metadata/client.go
@@ -42,10 +42,10 @@ import (
const defaultTimeout = "5s" // s
-func GetMetadataFromMetadataReport(revision string, instance
registry.ServiceInstance) (*info.MetadataInfo, error) {
- report := GetMetadataReport()
+func GetMetadataFromMetadataReport(revision string, instance
registry.ServiceInstance, registryId string) (*info.MetadataInfo, error) {
+ report := GetMetadataReportByRegistry(registryId)
if report == nil {
- return nil, perrors.New("no metadata report instance
found,please check ")
+ return nil, perrors.Errorf("no metadata report instance found
for registryId=%s, please check metadata-report configuration", registryId)
}
return report.GetAppMetadata(instance.GetServiceName(), revision)
}
diff --git a/metadata/client_test.go b/metadata/client_test.go
index 2f065f7b4..e5de35c97 100644
--- a/metadata/client_test.go
+++ b/metadata/client_test.go
@@ -35,6 +35,7 @@ import (
"dubbo.apache.org/dubbo-go/v3/common/constant"
"dubbo.apache.org/dubbo-go/v3/common/extension"
"dubbo.apache.org/dubbo-go/v3/metadata/info"
+ "dubbo.apache.org/dubbo-go/v3/metadata/report"
"dubbo.apache.org/dubbo-go/v3/protocol/base"
"dubbo.apache.org/dubbo-go/v3/protocol/result"
_ "dubbo.apache.org/dubbo-go/v3/proxy/proxy_factory"
@@ -62,22 +63,64 @@ var (
)
func TestGetMetadataFromMetadataReport(t *testing.T) {
+ t.Cleanup(func() { instances = make(map[string]report.MetadataReport) })
+
t.Run("no report instance", func(t *testing.T) {
- _, err := GetMetadataFromMetadataReport("1", ins)
+ instances = make(map[string]report.MetadataReport)
+ _, err := GetMetadataFromMetadataReport("1", ins, "default")
require.Error(t, err)
})
- mockReport := new(mockMetadataReport)
- defer mockReport.AssertExpectations(t)
- instances["default"] = mockReport
- t.Run("normal", func(t *testing.T) {
+
+ t.Run("default registry routes to default report", func(t *testing.T) {
+ instances = make(map[string]report.MetadataReport)
+ mockReport := new(mockMetadataReport)
+ defer mockReport.AssertExpectations(t)
+ instances["default"] = mockReport
+
mockReport.On("GetAppMetadata").Return(metadataInfo, nil).Once()
- got, err := GetMetadataFromMetadataReport("1", ins)
+ got, err := GetMetadataFromMetadataReport("1", ins, "default")
+ require.NoError(t, err)
+ assert.Equal(t, metadataInfo, got)
+ })
+
+ t.Run("specific registryId routes to its own report", func(t
*testing.T) {
+ instances = make(map[string]report.MetadataReport)
+ defaultReport := new(mockMetadataReport)
+ specificReport := new(mockMetadataReport)
+ defer defaultReport.AssertExpectations(t)
+ defer specificReport.AssertExpectations(t)
+ instances["default"] = defaultReport
+ instances["reg-a"] = specificReport
+
+ // specificReport must be called; defaultReport must NOT be
called
+ specificReport.On("GetAppMetadata").Return(metadataInfo,
nil).Once()
+ got, err := GetMetadataFromMetadataReport("1", ins, "reg-a")
+ require.NoError(t, err)
+ assert.Equal(t, metadataInfo, got)
+ })
+
+ t.Run("unknown registryId falls back to default report", func(t
*testing.T) {
+ instances = make(map[string]report.MetadataReport)
+ defaultReport := new(mockMetadataReport)
+ defer defaultReport.AssertExpectations(t)
+ instances["default"] = defaultReport
+
+ // When the specific registryId is not found, it falls back to
"default"
+ // so the default report's GetAppMetadata is called
+ defaultReport.On("GetAppMetadata").Return(metadataInfo,
nil).Once()
+ got, err := GetMetadataFromMetadataReport("1", ins,
"nonexistent-registry")
require.NoError(t, err)
assert.Equal(t, metadataInfo, got)
})
- t.Run("error", func(t *testing.T) {
+
+ t.Run("report error propagated", func(t *testing.T) {
+ instances = make(map[string]report.MetadataReport)
+ mockReport := new(mockMetadataReport)
+ defer mockReport.AssertExpectations(t)
+ instances["default"] = mockReport
+
mockReport.On("GetAppMetadata").Return(metadataInfo,
errors.New("mock error")).Once()
- _, err := GetMetadataFromMetadataReport("1", ins)
+ _, err := GetMetadataFromMetadataReport("1", ins, "default")
require.Error(t, err)
})
}
diff --git a/metadata/mapping/metadata/service_name_mapping.go
b/metadata/mapping/metadata/service_name_mapping.go
index 717864da5..c0f730ce7 100644
--- a/metadata/mapping/metadata/service_name_mapping.go
+++ b/metadata/mapping/metadata/service_name_mapping.go
@@ -119,18 +119,56 @@ func backoff(attempt int) time.Duration {
// Get will return the application-level services. If not found, the empty set
will be returned.
func (d *ServiceNameMapping) Get(url *common.URL, listener
mapping.MappingListener) (*gxset.HashSet, error) {
serviceInterface := url.GetParam(constant.InterfaceKey, "")
- metadataReport := metadata.GetMetadataReport()
- if metadataReport == nil {
+ metadataReports := metadata.GetMetadataReports()
+ if len(metadataReports) == 0 {
return nil, perrors.New("can not get mapping in remote cause no
metadata report instance found")
}
- return metadataReport.GetServiceAppMapping(serviceInterface,
DefaultGroup, listener)
+ // Attach the listener to the stable primary report only
(GetMetadataReport uses
+ // a deterministic selection: prefer "default", otherwise lexicographic
first).
+ // GetMetadataReports() iterates a map so its order is
non-deterministic; using
+ // i==0 as the anchor would bind the listener to a random backend each
run.
+ primaryReport := metadata.GetMetadataReport()
+ var result *gxset.HashSet
+ var errs []error
+ for _, metadataReport := range metadataReports {
+ var reportListener mapping.MappingListener
+ if metadataReport == primaryReport {
+ reportListener = listener
+ }
+ set, err :=
metadataReport.GetServiceAppMapping(serviceInterface, DefaultGroup,
reportListener)
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ if result == nil {
+ result = set
+ } else {
+ result.Add(set.Values()...)
+ }
+ }
+ if result == nil {
+ return nil, errors.Join(errs...)
+ }
+ return result, nil
}
+// Remove removes the service-to-app mapping for the given URL from all
+// registered metadata reports. Unlike Map (which stops on the first failure),
+// Remove is best-effort: it attempts every report and returns all errors
+// joined together so the caller can see the full failure picture. The
+// intent is to avoid leaving stale entries in any registry due to a transient
+// error in one of the others.
func (d *ServiceNameMapping) Remove(url *common.URL) error {
serviceInterface := url.GetParam(constant.InterfaceKey, "")
- metadataReport := metadata.GetMetadataReport()
- if metadataReport == nil {
+ metadataReports := metadata.GetMetadataReports()
+ if len(metadataReports) == 0 {
return perrors.New("can not remove mapping in remote cause no
metadata report instance found")
}
- return metadataReport.RemoveServiceAppMappingListener(serviceInterface,
DefaultGroup)
+ var errs []error
+ for _, metadataReport := range metadataReports {
+ if err :=
metadataReport.RemoveServiceAppMappingListener(serviceInterface, DefaultGroup);
err != nil {
+ errs = append(errs, err)
+ }
+ }
+ return errors.Join(errs...)
}
diff --git a/metadata/mapping/metadata/service_name_mapping_test.go
b/metadata/mapping/metadata/service_name_mapping_test.go
index c279ff4c0..fb40bc56d 100644
--- a/metadata/mapping/metadata/service_name_mapping_test.go
+++ b/metadata/mapping/metadata/service_name_mapping_test.go
@@ -19,6 +19,7 @@ package metadata
import (
"errors"
+ "sync"
"testing"
)
@@ -147,6 +148,105 @@ func initMock() (*mockMetadataReport, error) {
return metadataReport, err
}
+func initMockWithId(t *testing.T, registryId string) *mockMetadataReport {
+ t.Helper()
+ mockReport := new(mockMetadataReport)
+ extension.SetMetadataReportFactory(registryId, func()
report.MetadataReportFactory {
+ return mockReport
+ })
+ opts := metadata.NewReportOptions(
+ metadata.WithRegistryId(registryId),
+ metadata.WithProtocol(registryId),
+ metadata.WithAddress("127.0.0.1"),
+ )
+ require.NoError(t, opts.Init())
+ return mockReport
+}
+
+func TestServiceNameMappingRemoveFansOutToAllReports(t *testing.T) {
+ metadata.ClearMetadataReportInstances()
+ t.Cleanup(metadata.ClearMetadataReportInstances)
+ serviceNameMappingOnce = sync.Once{}
+ serviceNameMappingInstance = nil
+
+ r1 := initMockWithId(t, "reg-a")
+ r2 := initMockWithId(t, "reg-b")
+
+ ins := GetNameMappingInstance()
+ serviceUrl := common.NewURLWithOptions(
+ common.WithInterface("org.example.FooService"),
+ common.WithParamsValue(constant.ApplicationKey, "foo-app"),
+ )
+
+ r1.On("RemoveServiceAppMappingListener").Return(nil).Once()
+ r2.On("RemoveServiceAppMappingListener").Return(nil).Once()
+
+ err := ins.Remove(serviceUrl)
+ require.NoError(t, err)
+ r1.AssertExpectations(t)
+ r2.AssertExpectations(t)
+}
+
+func TestServiceNameMappingRemoveCollectsAllErrors(t *testing.T) {
+ metadata.ClearMetadataReportInstances()
+ t.Cleanup(metadata.ClearMetadataReportInstances)
+ serviceNameMappingOnce = sync.Once{}
+ serviceNameMappingInstance = nil
+
+ r1 := initMockWithId(t, "reg-c")
+ r2 := initMockWithId(t, "reg-d")
+
+ ins := GetNameMappingInstance()
+ serviceUrl := common.NewURLWithOptions(
+ common.WithInterface("org.example.BarService"),
+ common.WithParamsValue(constant.ApplicationKey, "bar-app"),
+ )
+
+ err1 := errors.New("r1 failure")
+ err2 := errors.New("r2 failure")
+
+ // both reports fail
+ r1.On("RemoveServiceAppMappingListener").Return(err1).Once()
+ r2.On("RemoveServiceAppMappingListener").Return(err2).Once()
+
+ err := ins.Remove(serviceUrl)
+ require.Error(t, err)
+ // both individual errors must be present in the returned error
+ require.ErrorIs(t, err, err1)
+ require.ErrorIs(t, err, err2)
+ r1.AssertExpectations(t)
+ r2.AssertExpectations(t)
+}
+
+func TestServiceNameMappingRemoveContinuesAfterPartialFailure(t *testing.T) {
+ metadata.ClearMetadataReportInstances()
+ t.Cleanup(metadata.ClearMetadataReportInstances)
+ serviceNameMappingOnce = sync.Once{}
+ serviceNameMappingInstance = nil
+
+ r1 := initMockWithId(t, "reg-e")
+ r2 := initMockWithId(t, "reg-f")
+
+ ins := GetNameMappingInstance()
+ serviceUrl := common.NewURLWithOptions(
+ common.WithInterface("org.example.BazService"),
+ common.WithParamsValue(constant.ApplicationKey, "baz-app"),
+ )
+
+ removeErr := errors.New("r1 partial failure")
+
+ // r1 fails, r2 succeeds — the loop must not short-circuit
+ r1.On("RemoveServiceAppMappingListener").Return(removeErr).Once()
+ r2.On("RemoveServiceAppMappingListener").Return(nil).Once()
+
+ err := ins.Remove(serviceUrl)
+ // the joined error only contains r1's error; the call should error
+ require.ErrorIs(t, err, removeErr)
+ // both reports must have been called despite r1's failure
+ r1.AssertExpectations(t)
+ r2.AssertExpectations(t)
+}
+
type listener struct {
}
diff --git a/metadata/report_instance.go b/metadata/report_instance.go
index f1157ca18..55cf516df 100644
--- a/metadata/report_instance.go
+++ b/metadata/report_instance.go
@@ -18,11 +18,12 @@
package metadata
import (
+ "sort"
"time"
)
import (
- "github.com/dubbogo/gost/container/set"
+ gxset "github.com/dubbogo/gost/container/set"
"github.com/dubbogo/gost/log/logger"
)
@@ -41,6 +42,12 @@ var (
instances = make(map[string]report.MetadataReport)
)
+// ClearMetadataReportInstances resets the package-level instances map.
+// Intended for test isolation only; do not call in production code.
+func ClearMetadataReportInstances() {
+ instances = make(map[string]report.MetadataReport)
+}
+
func addMetadataReport(registryId string, url *common.URL) error {
fac := extension.GetMetadataReportFactory(url.Protocol)
if fac == nil {
@@ -51,21 +58,46 @@ func addMetadataReport(registryId string, url *common.URL)
error {
return nil
}
+// GetMetadataReport returns a single metadata report for callers that lack
+// registry context. It prefers the "default" registry's report; when absent
+// it falls back to the lexicographically first registry id so the selection
+// is always stable across calls.
func GetMetadataReport() report.MetadataReport {
- for _, v := range instances {
- return v
+ if r, ok := instances[constant.DefaultKey]; ok {
+ return r
+ }
+ keys := make([]string, 0, len(instances))
+ for k := range instances {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ if len(keys) > 0 {
+ return instances[keys[0]]
}
return nil
}
+// GetMetadataReportByRegistry returns the metadata report bound to the given
+// registry id. When the registry id is empty the caller has no registry
context,
+// so the stable default returned by GetMetadataReport is used. When a specific
+// (non-empty) registry id is not found, it falls back to the "default" report
+// if one exists. This handles the common case where a standalone
metadata-report
+// config is registered under "default" while named registries (e.g. nacos, zk)
+// need to use it. nil is returned only when neither the specific id nor
"default"
+// is registered.
func GetMetadataReportByRegistry(registry string) report.MetadataReport {
if len(registry) == 0 {
- registry = constant.DefaultKey
+ return GetMetadataReport()
}
if r, ok := instances[registry]; ok {
return r
}
- return GetMetadataReport()
+ if r, ok := instances[constant.DefaultKey]; ok {
+ logger.Infof("[Metadata] no metadata report bound to
registryId=%s, falling back to default", registry)
+ return r
+ }
+ logger.Warnf("[Metadata] no metadata report found for registryId=%s",
registry)
+ return nil
}
func GetMetadataReports() []report.MetadataReport {
diff --git a/metadata/report_instance_test.go b/metadata/report_instance_test.go
index 88a631a54..dfe31b78c 100644
--- a/metadata/report_instance_test.go
+++ b/metadata/report_instance_test.go
@@ -176,13 +176,61 @@ func TestGetMetadataReport(t *testing.T) {
assert.NotNil(t, GetMetadataReport())
}
+func TestGetMetadataReportIsDeterministic(t *testing.T) {
+ instances = make(map[string]report.MetadataReport)
+ r1 := new(mockMetadataReport)
+ r2 := new(mockMetadataReport)
+ // "aaa" sorts before "zzz" alphabetically, neither is "default"
+ instances["zzz"] = r1
+ instances["aaa"] = r2
+
+ // without a "default" key, must always return the alphabetically first
entry (r2)
+ for range 20 {
+ got := GetMetadataReport()
+ assert.Equal(t, r2, got, "expected the report for 'aaa'")
+ }
+
+ // when the "default" key exists, must always return it (r1), taking
priority over alphabetical order
+ instances[constant.DefaultKey] = r1
+ for range 20 {
+ got := GetMetadataReport()
+ assert.Equal(t, r1, got, "expected the report for 'default'")
+ }
+}
+
func TestGetMetadataReportByRegistry(t *testing.T) {
instances = make(map[string]report.MetadataReport)
+ // nothing registered: all paths return nil
+ assert.Nil(t, GetMetadataReportByRegistry(""))
assert.Nil(t, GetMetadataReportByRegistry("reg"))
- instances["default"] = new(mockMetadataReport)
- assert.NotNil(t, GetMetadataReportByRegistry("default"))
- assert.NotNil(t, GetMetadataReportByRegistry("reg"))
- assert.NotNil(t, GetMetadataReportByRegistry(""))
+
+ defaultReport := new(mockMetadataReport)
+ instances["default"] = defaultReport
+
+ // exact hit
+ assert.Equal(t, defaultReport, GetMetadataReportByRegistry("default"))
+ // empty string → no registry context → falls through to
GetMetadataReport() → "default"
+ assert.Equal(t, defaultReport, GetMetadataReportByRegistry(""))
+ // specific but unknown id → falls back to "default"
+ assert.Equal(t, defaultReport, GetMetadataReportByRegistry("reg"))
+}
+
+func TestGetMetadataReportByRegistryFallsBackDeterministically(t *testing.T) {
+ instances = make(map[string]report.MetadataReport)
+ rA := new(mockMetadataReport)
+ rB := new(mockMetadataReport)
+ instances["aaa"] = rA // lex-first
+ instances["zzz"] = rB
+
+ // known key → exact report
+ assert.Equal(t, rA, GetMetadataReportByRegistry("aaa"))
+ assert.Equal(t, rB, GetMetadataReportByRegistry("zzz"))
+
+ // unknown specific id → nil when no "default" is registered
+ assert.Nil(t, GetMetadataReportByRegistry("unknown-registry"))
+
+ // empty string → falls through to GetMetadataReport() → lex-first
("aaa" → rA)
+ assert.Equal(t, rA, GetMetadataReportByRegistry(""))
}
func TestGetMetadataReports(t *testing.T) {
diff --git a/registry/nacos/service_discovery_test.go
b/registry/nacos/service_discovery_test.go
index 1c068a2bc..5ee5ff201 100644
--- a/registry/nacos/service_discovery_test.go
+++ b/registry/nacos/service_discovery_test.go
@@ -150,7 +150,7 @@ func TestFunction(t *testing.T) {
hs := gxset.NewSet()
hs.Add(testName)
- sicl := servicediscovery.NewServiceInstancesChangedListener("test_app",
hs)
+ sicl := servicediscovery.NewServiceInstancesChangedListener("test_app",
constant.DefaultKey, hs)
sicl.AddListenerAndNotify(testName, tn)
err = sd.AddListener(sicl)
require.NoError(t, err)
diff --git
a/registry/servicediscovery/customizer/metadata_service_url_params_customizer.go
b/registry/servicediscovery/customizer/metadata_service_url_params_customizer.go
index eee5645af..45e558d06 100644
---
a/registry/servicediscovery/customizer/metadata_service_url_params_customizer.go
+++
b/registry/servicediscovery/customizer/metadata_service_url_params_customizer.go
@@ -56,7 +56,10 @@ func (m *metadataServiceURLParamsMetadataCustomizer)
GetPriority() int {
}
func (m *metadataServiceURLParamsMetadataCustomizer) Customize(instance
registry.ServiceInstance) {
- //todo Multi-instance metadata alignment needs to be improved
+ // TODO: GetMetadataService() is a global singleton and returns the
same metadata service URL
+ // regardless of which registry this instance belongs to. In a
multi-registry setup each
+ // registry should expose its own metadata service URL. This requires
per-registry metadata
+ // service tracking and is left for a follow-up.
url, _ := metadata.GetMetadataService().GetMetadataServiceURL()
if url == nil {
// when metadata service is not exported the url will be
nil,this is because metadata type is remote
diff --git
a/registry/servicediscovery/customizer/service_revision_customizer.go
b/registry/servicediscovery/customizer/service_revision_customizer.go
index a9598d997..e3b220da7 100644
--- a/registry/servicediscovery/customizer/service_revision_customizer.go
+++ b/registry/servicediscovery/customizer/service_revision_customizer.go
@@ -49,12 +49,20 @@ func (e *exportedServicesRevisionMetadataCustomizer)
GetPriority() int {
return 1
}
-// Customize calculate the revision for exported urls and then put it into
instance metadata
+// Customize calculates the revision of exported services scoped to the
registry,
+// preventing different instances from getting the same revision due to a
merged cross-registry service list in multi-registry setups.
func (e *exportedServicesRevisionMetadataCustomizer) Customize(instance
registry.ServiceInstance) {
- urls, err := metadata.GetMetadataService().GetExportedServiceURLs()
- if err != nil {
- logger.Errorf("[Registry][ServiceDiscovery] get metadata
service url is error, err=%v", err)
- return
+ registryId := instance.GetMetadata()[constant.RegistryIdKey]
+ if len(registryId) == 0 {
+ // revision will be "0" (no services found for empty key),
which causes OnEvent to skip
+ // this instance entirely — ensure RegistryIdKey is set before
customizers run.
+ logger.Errorf("[Registry][ServiceDiscovery] instance has no
registryId in metadata; " +
+ "exported revision will be \"0\" and this instance will
be invisible to consumers")
+ }
+ metaInfo := metadata.GetMetadataInfo(registryId)
+ var urls []*common.URL
+ if metaInfo != nil {
+ urls = metaInfo.GetExportedServiceURLs()
}
revision := resolveRevision(urls)
if len(revision) == 0 {
@@ -70,12 +78,19 @@ func (e *subscribedServicesRevisionMetadataCustomizer)
GetPriority() int {
return 2
}
-// Customize calculate the revision for subscribed urls and then put it into
instance metadata
+// Customize calculates the revision of subscribed services scoped to the
registry.
func (e *subscribedServicesRevisionMetadataCustomizer) Customize(instance
registry.ServiceInstance) {
- urls, err := metadata.GetMetadataService().GetSubscribedURLs()
- if err != nil {
- logger.Errorf("[Registry][ServiceDiscovery] get metadata
subscribed url is error, err=%v", err)
- return
+ registryId := instance.GetMetadata()[constant.RegistryIdKey]
+ if len(registryId) == 0 {
+ // revision will be "0" (no subscriptions found for empty key),
which causes OnEvent to skip
+ // this instance entirely — ensure RegistryIdKey is set before
customizers run.
+ logger.Errorf("[Registry][ServiceDiscovery] instance has no
registryId in metadata; " +
+ "subscribed revision will be \"0\" and this instance
will be invisible to consumers")
+ }
+ metaInfo := metadata.GetMetadataInfo(registryId)
+ var urls []*common.URL
+ if metaInfo != nil {
+ urls = metaInfo.GetSubscribedURLs()
}
revision := resolveRevision(urls)
if len(revision) == 0 {
diff --git
a/registry/servicediscovery/customizer/service_revision_customizer_test.go
b/registry/servicediscovery/customizer/service_revision_customizer_test.go
new file mode 100644
index 000000000..8ecaa763e
--- /dev/null
+++ b/registry/servicediscovery/customizer/service_revision_customizer_test.go
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package customizer
+
+import (
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/metadata"
+ "dubbo.apache.org/dubbo-go/v3/registry"
+)
+
+// TestExportedRevisionIsRegistryScoped verifies that when two registries
export different
+// service sets, their instances get different revision values — not a merged
cross-registry one.
+func TestExportedRevisionIsRegistryScoped(t *testing.T) {
+ metadata.ClearMetadataReportInstances()
+ t.Cleanup(func() {
+ metadata.ClearMetadataReportInstances()
+ })
+
+ urlA := common.NewURLWithOptions(
+ common.WithInterface("org.example.ServiceA"),
+ common.WithParamsValue(constant.ApplicationKey, "app"),
+ common.WithPort("20880"),
+ )
+ urlB := common.NewURLWithOptions(
+ common.WithInterface("org.example.ServiceB"),
+ common.WithParamsValue(constant.ApplicationKey, "app"),
+ common.WithPort("20881"),
+ )
+
+ metadata.AddService("rev-reg-a", urlA)
+ metadata.AddService("rev-reg-b", urlB)
+ t.Cleanup(func() {
+ metadata.RemoveService("rev-reg-a", urlA)
+ metadata.RemoveService("rev-reg-b", urlB)
+ })
+
+ cus := &exportedServicesRevisionMetadataCustomizer{}
+
+ instA := ®istry.DefaultServiceInstance{
+ Metadata: map[string]string{constant.RegistryIdKey:
"rev-reg-a"},
+ }
+ cus.Customize(instA)
+ revA :=
instA.GetMetadata()[constant.ExportedServicesRevisionPropertyName]
+
+ instB := ®istry.DefaultServiceInstance{
+ Metadata: map[string]string{constant.RegistryIdKey:
"rev-reg-b"},
+ }
+ cus.Customize(instB)
+ revB :=
instB.GetMetadata()[constant.ExportedServicesRevisionPropertyName]
+
+ assert.NotEqual(t, revA, revB, "different registries with different
services should produce different revisions")
+ assert.NotEqual(t, "0", revA, "reg-a has a service, revision should not
be 0")
+ assert.NotEqual(t, "0", revB, "reg-b has a service, revision should not
be 0")
+}
+
+// TestExportedRevisionMissingRegistryIdYieldsZero verifies that an instance
with no
+// RegistryIdKey gets revision "0" (no services found), and does not panic or
use
+// another registry's service list.
+func TestExportedRevisionMissingRegistryIdYieldsZero(t *testing.T) {
+ metadata.ClearMetadataReportInstances()
+ t.Cleanup(metadata.ClearMetadataReportInstances)
+
+ // Register a service under a real registry so we can confirm it is NOT
used
+ urlA := common.NewURLWithOptions(
+ common.WithInterface("org.example.ShouldBeIsolated"),
+ common.WithParamsValue(constant.ApplicationKey, "app"),
+ common.WithPort("20880"),
+ )
+ metadata.AddService("some-registry", urlA)
+ t.Cleanup(func() {
+ metadata.RemoveService("some-registry", urlA)
+ })
+
+ cus := &exportedServicesRevisionMetadataCustomizer{}
+
+ inst := ®istry.DefaultServiceInstance{
+ Metadata: map[string]string{}, // no RegistryIdKey
+ }
+ cus.Customize(inst)
+ rev := inst.GetMetadata()[constant.ExportedServicesRevisionPropertyName]
+
+ // GetMetadataInfo("") returns nil → resolveRevision(nil) == "0"
+ assert.Equal(t, "0", rev, "instance with no registryId should get
revision 0, not borrow another registry's service list")
+}
+
+// TestSubscribedRevisionIsRegistryScoped mirrors the exported test for
subscribed URLs.
+func TestSubscribedRevisionIsRegistryScoped(t *testing.T) {
+ metadata.ClearMetadataReportInstances()
+ t.Cleanup(metadata.ClearMetadataReportInstances)
+
+ urlA := common.NewURLWithOptions(
+ common.WithInterface("org.example.SubA"),
+ common.WithParamsValue(constant.ApplicationKey, "app"),
+ common.WithPort("20880"),
+ )
+ urlB := common.NewURLWithOptions(
+ common.WithInterface("org.example.SubB"),
+ common.WithParamsValue(constant.ApplicationKey, "app"),
+ common.WithPort("20881"),
+ )
+
+ metadata.AddSubscribeURL("sub-reg-a", urlA)
+ metadata.AddSubscribeURL("sub-reg-b", urlB)
+ t.Cleanup(func() {
+ metadata.RemoveSubscribeURL("sub-reg-a", urlA)
+ metadata.RemoveSubscribeURL("sub-reg-b", urlB)
+ })
+
+ cus := &subscribedServicesRevisionMetadataCustomizer{}
+
+ instA := ®istry.DefaultServiceInstance{
+ Metadata: map[string]string{constant.RegistryIdKey:
"sub-reg-a"},
+ }
+ cus.Customize(instA)
+ revA :=
instA.GetMetadata()[constant.SubscribedServicesRevisionPropertyName]
+
+ instB := ®istry.DefaultServiceInstance{
+ Metadata: map[string]string{constant.RegistryIdKey:
"sub-reg-b"},
+ }
+ cus.Customize(instB)
+ revB :=
instB.GetMetadata()[constant.SubscribedServicesRevisionPropertyName]
+
+ assert.NotEqual(t, revA, revB, "different registries with different
subscriptions should produce different revisions")
+}
+
+// TestSubscribedRevisionMissingRegistryIdYieldsZero mirrors the exported
missing-key test.
+func TestSubscribedRevisionMissingRegistryIdYieldsZero(t *testing.T) {
+ metadata.ClearMetadataReportInstances()
+ t.Cleanup(metadata.ClearMetadataReportInstances)
+
+ urlA := common.NewURLWithOptions(
+ common.WithInterface("org.example.SubIsolated"),
+ common.WithParamsValue(constant.ApplicationKey, "app"),
+ common.WithPort("20880"),
+ )
+ metadata.AddSubscribeURL("some-registry", urlA)
+ t.Cleanup(func() {
+ metadata.RemoveSubscribeURL("some-registry", urlA)
+ })
+
+ cus := &subscribedServicesRevisionMetadataCustomizer{}
+ inst := ®istry.DefaultServiceInstance{
+ Metadata: map[string]string{},
+ }
+ cus.Customize(inst)
+ rev :=
inst.GetMetadata()[constant.SubscribedServicesRevisionPropertyName]
+
+ assert.Equal(t, "0", rev, "instance with no registryId should get
revision 0, not borrow another registry's subscriptions")
+}
diff --git a/registry/servicediscovery/service_discovery_registry.go
b/registry/servicediscovery/service_discovery_registry.go
index 6d07dd318..e01d42154 100644
--- a/registry/servicediscovery/service_discovery_registry.go
+++ b/registry/servicediscovery/service_discovery_registry.go
@@ -86,13 +86,14 @@ func newServiceDiscoveryRegistry(url *common.URL)
(registry.Registry, error) {
}
func (s *serviceDiscoveryRegistry) RegisterService() error {
- metaInfo :=
metadata.GetMetadataInfo(s.url.GetParam(constant.RegistryIdKey, ""))
+ registryId := s.url.GetParam(constant.RegistryIdKey,
constant.DefaultKey)
+ metaInfo := metadata.GetMetadataInfo(registryId)
if metaInfo == nil {
- panic("no metada info found of registry id " +
s.url.GetParam(constant.RegistryIdKey, ""))
+ panic("no metada info found of registry id " + registryId)
}
urls := metaInfo.GetExportedServiceURLs()
for _, url := range urls {
- instance := createInstance(metaInfo, url)
+ instance := createInstance(metaInfo, url, registryId)
metaInfo.Revision =
instance.GetMetadata()[constant.ExportedServicesRevisionPropertyName]
if metadata.GetMetadataType() ==
constant.RemoteMetadataStorageType {
if s.metadataReport == nil {
@@ -115,9 +116,12 @@ func (s *serviceDiscoveryRegistry) RegisterService() error
{
return nil
}
-func createInstance(meta *info.MetadataInfo, url *common.URL)
registry.ServiceInstance {
+func createInstance(meta *info.MetadataInfo, url *common.URL, registryId
string) registry.ServiceInstance {
params := make(map[string]string, 8)
params[constant.MetadataStorageTypePropertyName] =
metadata.GetMetadataType()
+ // Expose the registry this instance belongs to so that customizers
(e.g. revision
+ // calculators) can scope their work to the correct per-registry
service set.
+ params[constant.RegistryIdKey] = registryId
// Keep routing attributes visible on the registered instance as well
as in service metadata.
if environment := url.GetParam(constant.EnvironmentKey, "");
len(environment) > 0 {
params[constant.EnvironmentKey] = environment
@@ -212,11 +216,11 @@ func (s *serviceDiscoveryRegistry) UnSubscribe(url
*common.URL, listener registr
}
func (s *serviceDiscoveryRegistry)
syncExportedMetadataAfterUnregister(targetURL *common.URL, origin
[]registry.ServiceInstance, keep []registry.ServiceInstance) error {
- registryID, exist := s.url.GetNonDefaultParam(constant.RegistryIdKey)
+ registryId, exist := s.url.GetNonDefaultParam(constant.RegistryIdKey)
if !exist {
return nil
}
- metadataInfo := metadata.GetMetadataInfo(registryID)
+ metadataInfo := metadata.GetMetadataInfo(registryId)
if metadataInfo == nil {
return nil
}
@@ -225,14 +229,14 @@ func (s *serviceDiscoveryRegistry)
syncExportedMetadataAfterUnregister(targetURL
if len(origin) > 0 {
metadataInfo.ReplaceExportedServices(keepURLs)
} else if targetURL != nil {
- metadata.RemoveService(registryID, targetURL)
+ metadata.RemoveService(registryId, targetURL)
}
remainingURLs := metadataInfo.GetExportedServiceURLs()
if len(remainingURLs) == 0 {
metadataInfo.Revision = "0"
return nil
}
- instance := createInstance(metadataInfo, remainingURLs[0])
+ instance := createInstance(metadataInfo, remainingURLs[0], registryId)
revision :=
instance.GetMetadata()[constant.ExportedServicesRevisionPropertyName]
metadataInfo.Revision = revision
if len(keepURLs) == 0 {
@@ -360,7 +364,7 @@ func (s *serviceDiscoveryRegistry) SubscribeURL(url
*common.URL, notify registry
protocolServiceKey := url.ServiceKey() + ":" + protocol
listener := s.serviceListeners[serviceNamesKey]
if listener == nil {
- listener =
NewServiceInstancesChangedListener(url.GetParam(constant.ApplicationKey, ""),
services)
+ listener =
NewServiceInstancesChangedListener(url.GetParam(constant.ApplicationKey, ""),
s.url.GetParam(constant.RegistryIdKey, constant.DefaultKey), services)
for _, serviceNameTmp := range services.Values() {
serviceName := serviceNameTmp.(string)
instances :=
s.serviceDiscovery.GetInstances(serviceName)
diff --git a/registry/servicediscovery/service_discovery_registry_test.go
b/registry/servicediscovery/service_discovery_registry_test.go
index 46f295705..77c83b7d6 100644
--- a/registry/servicediscovery/service_discovery_registry_test.go
+++ b/registry/servicediscovery/service_discovery_registry_test.go
@@ -57,7 +57,7 @@ const (
// TestServiceDiscoveryRegistryRegister verifies the registration process.
func TestServiceDiscoveryRegistryRegister(t *testing.T) {
mockSD, mockMapping := setupEnvironment(t)
- regID := fmt.Sprintf("mock-reg-%d", time.Now().UnixNano())
+ regID := fmt.Sprintf("mock-reg-%s-%d", t.Name(), time.Now().UnixNano())
registryURL, err := common.NewURL(testRegistryURL,
common.WithParamsValue(constant.RegistryKey, "mock"),
@@ -127,7 +127,7 @@ func TestServiceDiscoveryRegistrySubscribe(t *testing.T) {
func TestServiceDiscoveryRegistryUnSubscribe(t *testing.T) {
mockSD, mockMapping := setupEnvironment(t)
mockMapping.data[testInterface] = gxset.NewSet(testApp)
- regID := fmt.Sprintf("mock-reg-%d", time.Now().UnixNano())
+ regID := fmt.Sprintf("mock-reg-%s-%d", t.Name(), time.Now().UnixNano())
registryURL, _ := common.NewURL(testRegistryURL,
common.WithParamsValue(constant.RegistryKey, "mock"),
@@ -159,7 +159,7 @@ func
TestServiceDiscoveryRegistryUnSubscribeKeepsMetadataOnRemoveFailure(t *test
mockSD, mockMapping := setupEnvironment(t)
mockMapping.data[testInterface] = gxset.NewSet(testApp)
mockMapping.removeErr = errors.New("mock remove failed")
- regID := fmt.Sprintf("mock-reg-%d", time.Now().UnixNano())
+ regID := fmt.Sprintf("mock-reg-%s-%d", t.Name(), time.Now().UnixNano())
registryURL, _ := common.NewURL(testRegistryURL,
common.WithParamsValue(constant.RegistryKey, "mock"),
@@ -188,7 +188,7 @@ func
TestServiceDiscoveryRegistryUnSubscribeKeepsMetadataOnRemoveFailure(t *test
func TestServiceDiscoveryRegistryUnRegisterSyncsBulkMetadataCleanup(t
*testing.T) {
mockSD, mockMapping := setupEnvironment(t)
- regID := fmt.Sprintf("mock-reg-%d", time.Now().UnixNano())
+ regID := fmt.Sprintf("mock-reg-%s-%d", t.Name(), time.Now().UnixNano())
registryURL, err := common.NewURL(testRegistryURL,
common.WithParamsValue(constant.RegistryKey, "mock"),
@@ -244,7 +244,7 @@ func
TestServiceDiscoveryRegistryUnRegisterSyncsBulkMetadataCleanup(t *testing.T
func TestServiceDiscoveryRegistryUnRegisterServicePartialFailSyncsMetadata(t
*testing.T) {
mockSD, mockMapping := setupEnvironment(t)
- regID := fmt.Sprintf("mock-reg-%d", time.Now().UnixNano())
+ regID := fmt.Sprintf("mock-reg-%s-%d", t.Name(), time.Now().UnixNano())
registryURL, err := common.NewURL(testRegistryURL,
common.WithParamsValue(constant.RegistryKey, "mock"),
@@ -294,7 +294,7 @@ func
TestServiceDiscoveryRegistryUnRegisterServicePartialFailSyncsMetadata(t *te
assert.Equal(t, providerURL1, metaInfo.GetExportedServiceURLs()[0])
assert.Len(t, metaInfo.Services, 1)
- expectedRevision := createInstance(metaInfo,
providerURL1).GetMetadata()[constant.ExportedServicesRevisionPropertyName]
+ expectedRevision := createInstance(metaInfo, providerURL1,
regID).GetMetadata()[constant.ExportedServicesRevisionPropertyName]
assert.Equal(t, expectedRevision, metaInfo.Revision)
assert.True(t, mockSD.updateCalled)
assert.Contains(t, mockSD.updatedIDs, providerURL1.Address())
@@ -302,7 +302,7 @@ func
TestServiceDiscoveryRegistryUnRegisterServicePartialFailSyncsMetadata(t *te
func
TestServiceDiscoveryRegistryUnRegisterWithoutTrackedInstancesReconcilesMetadata(t
*testing.T) {
_, mockMapping := setupEnvironment(t)
- regID := fmt.Sprintf("mock-reg-%d", time.Now().UnixNano())
+ regID := fmt.Sprintf("mock-reg-%s-%d", t.Name(), time.Now().UnixNano())
registryURL, err := common.NewURL(testRegistryURL,
common.WithParamsValue(constant.RegistryKey, "mock"),
@@ -344,7 +344,7 @@ func
TestServiceDiscoveryRegistryUnRegisterWithoutTrackedInstancesReconcilesMeta
assert.Equal(t, providerURL2, metaInfo.GetExportedServiceURLs()[0])
assert.Len(t, metaInfo.Services, 1)
- expectedRevision := createInstance(metaInfo,
providerURL2).GetMetadata()[constant.ExportedServicesRevisionPropertyName]
+ expectedRevision := createInstance(metaInfo, providerURL2,
regID).GetMetadata()[constant.ExportedServicesRevisionPropertyName]
assert.Equal(t, expectedRevision, metaInfo.Revision)
}
diff --git
a/registry/servicediscovery/service_instances_changed_listener_impl.go
b/registry/servicediscovery/service_instances_changed_listener_impl.go
index 13030d55d..af18900b2 100644
--- a/registry/servicediscovery/service_instances_changed_listener_impl.go
+++ b/registry/servicediscovery/service_instances_changed_listener_impl.go
@@ -58,6 +58,7 @@ func initCache(app string) {
// ServiceInstancesChangedListenerImpl The Service Discovery Changed Event
Listener
type ServiceInstancesChangedListenerImpl struct {
app string
+ registryId string
serviceNames *gxset.HashSet
listeners map[string]registry.NotifyListener
serviceUrls map[string][]*common.URL
@@ -66,12 +67,13 @@ type ServiceInstancesChangedListenerImpl struct {
mutex sync.Mutex
}
-func NewServiceInstancesChangedListener(app string, services *gxset.HashSet)
registry.ServiceInstancesChangedListener {
+func NewServiceInstancesChangedListener(app string, registryId string,
services *gxset.HashSet) registry.ServiceInstancesChangedListener {
cacheOnce.Do(func() {
initCache(app)
})
return &ServiceInstancesChangedListenerImpl{
app: app,
+ registryId: registryId,
serviceNames: services,
listeners: make(map[string]registry.NotifyListener),
serviceUrls: make(map[string][]*common.URL),
@@ -118,7 +120,7 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
revisionToInstances[revision] = append(subInstances,
instance)
metadataInfo := lstn.revisionToMetadata[revision]
if metadataInfo == nil {
- meta, err := GetMetadataInfo(lstn.app,
instance, revision)
+ meta, err := GetMetadataInfo(lstn.app,
instance, revision, lstn.registryId)
if err != nil {
// Skip this instance if metadata fetch
fails (e.g., old Java Dubbo version)
// Try next instance with same revision
@@ -147,7 +149,8 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
}
lstn.revisionToMetadata = newRevisionToMetadata
for revision, metadataInfo := range newRevisionToMetadata {
- metaCache.Set(revision, metadataInfo)
+ cacheKey := lstn.app + ":" + lstn.registryId + ":" + revision
+ metaCache.Set(cacheKey, metadataInfo)
}
for serviceKey, revisionServices := range serviceToRevisionServices {
@@ -242,11 +245,12 @@ func (lstn *ServiceInstancesChangedListenerImpl)
GetEventType() reflect.Type {
}
// GetMetadataInfo get metadata info when MetadataStorageTypePropertyName is
null
-func GetMetadataInfo(app string, instance registry.ServiceInstance, revision
string) (*info.MetadataInfo, error) {
+func GetMetadataInfo(app string, instance registry.ServiceInstance, revision
string, registryId string) (*info.MetadataInfo, error) {
cacheOnce.Do(func() {
initCache(app)
})
- if metadataInfo, ok := metaCache.Get(revision); ok {
+ cacheKey := app + ":" + registryId + ":" + revision
+ if metadataInfo, ok := metaCache.Get(cacheKey); ok {
return metadataInfo.(*info.MetadataInfo), nil
}
@@ -259,7 +263,7 @@ func GetMetadataInfo(app string, instance
registry.ServiceInstance, revision str
metadataStorageType =
instance.GetMetadata()[constant.MetadataStorageTypePropertyName]
}
if metadataStorageType == constant.RemoteMetadataStorageType {
- metadataInfo, err =
metadata.GetMetadataFromMetadataReport(revision, instance)
+ metadataInfo, err =
metadata.GetMetadataFromMetadataReport(revision, instance, registryId)
if err != nil {
return nil, err
}
@@ -269,6 +273,6 @@ func GetMetadataInfo(app string, instance
registry.ServiceInstance, revision str
return nil, err
}
}
- metaCache.Set(revision, metadataInfo)
+ metaCache.Set(cacheKey, metadataInfo)
return metadataInfo, nil
}
diff --git
a/registry/servicediscovery/service_instances_changed_listener_impl_test.go
b/registry/servicediscovery/service_instances_changed_listener_impl_test.go
index a289de37c..24a09a642 100644
--- a/registry/servicediscovery/service_instances_changed_listener_impl_test.go
+++ b/registry/servicediscovery/service_instances_changed_listener_impl_test.go
@@ -26,18 +26,23 @@ import (
gxset "github.com/dubbogo/gost/container/set"
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
import (
"dubbo.apache.org/dubbo-go/v3/common"
"dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/metadata"
"dubbo.apache.org/dubbo-go/v3/metadata/info"
+ "dubbo.apache.org/dubbo-go/v3/metadata/mapping"
+ metadatareport "dubbo.apache.org/dubbo-go/v3/metadata/report"
"dubbo.apache.org/dubbo-go/v3/registry"
)
func TestServiceInstancesChangedListenerAggregatesSameServiceAcrossRevisions(t
*testing.T) {
- listener := NewServiceInstancesChangedListener(testApp,
gxset.NewSet(testApp))
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey, gxset.NewSet(testApp))
notify := &capturingNotifyListener{}
listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
@@ -52,7 +57,7 @@ func
TestServiceInstancesChangedListenerAggregatesSameServiceAcrossRevisions(t *
}
func
TestServiceInstancesChangedListenerRefreshesURLsOnProviderRemoveAndRestart(t
*testing.T) {
- listener := NewServiceInstancesChangedListener(testApp,
gxset.NewSet(testApp))
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey, gxset.NewSet(testApp))
notify := &capturingNotifyListener{}
listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
@@ -83,12 +88,12 @@ func
TestServiceInstancesChangedListenerRefreshesURLsOnProviderRemoveAndRestart(
}
func
TestServiceInstancesChangedListenerRefreshesAndClearsEnvironmentWhenRevisionIsUnchanged(t
*testing.T) {
- listener := NewServiceInstancesChangedListener(testApp,
gxset.NewSet(testApp))
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey, gxset.NewSet(testApp))
notify := &capturingNotifyListener{}
listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
revision := "rev-20001-same-environment-change"
- metaCache.Set(revision, newTestMetadataInfo(t, revision, 20001, "pre"))
+ metaCache.Set(testApp+":"+constant.DefaultKey+":"+revision,
newTestMetadataInfo(t, revision, 20001, "pre"))
pre := newTestServiceInstanceOnly(20001, "pre", revision)
require.NoError(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{
@@ -119,15 +124,16 @@ func
TestServiceInstancesChangedListenerRefreshesAndClearsEnvironmentWhenRevisio
}
func TestServiceInstancesChangedListenerSkipsNilMetadataWithoutPanic(t
*testing.T) {
- listener := NewServiceInstancesChangedListener(testApp,
gxset.NewSet(testApp))
+ listener := NewServiceInstancesChangedListener(testApp,
constant.DefaultKey, gxset.NewSet(testApp))
notify := &capturingNotifyListener{}
listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
revision := "rev-20003-nil-metadata"
var metadataInfo *info.MetadataInfo
- metaCache.Set(revision, metadataInfo)
+ cacheKey := testApp + ":" + constant.DefaultKey + ":" + revision
+ metaCache.Set(cacheKey, metadataInfo)
t.Cleanup(func() {
- metaCache.Delete(revision)
+ metaCache.Delete(cacheKey)
})
instance := newTestServiceInstanceOnly(20003, "pre", revision)
@@ -145,11 +151,114 @@ func TestCreateInstanceCarriesEnvironmentMetadata(t
*testing.T) {
meta := info.NewMetadataInfo(testApp, "")
providerURL := newTestProviderURL(t, 20001, "pre")
- instance := createInstance(meta, providerURL)
+ instance := createInstance(meta, providerURL, constant.DefaultKey)
assert.Equal(t, "pre", instance.GetMetadata()[constant.EnvironmentKey])
}
+// TestListenerUsesRegistryIdToFetchRemoteMetadata verifies that when a
listener is
+// created with a specific registryId, GetMetadataInfo threads that id through
to
+// GetMetadataFromMetadataReport so the correct per-registry metadata report
is used.
+func TestListenerUsesRegistryIdToFetchRemoteMetadata(t *testing.T) {
+ const listenerRegistryId = "remote-reg-test"
+ const revision = "rev-remote-reg-test"
+
+ // Register a mock metadata report under the listener's registry id
+ mockReport := new(listenerMockMetadataReport)
+ extension.SetMetadataReportFactory(listenerRegistryId, func()
metadatareport.MetadataReportFactory {
+ return mockReport
+ })
+ opts := metadata.NewReportOptions(
+ metadata.WithRegistryId(listenerRegistryId),
+ metadata.WithProtocol(listenerRegistryId),
+ metadata.WithAddress("127.0.0.1"),
+ )
+ require.NoError(t, opts.Init())
+ t.Cleanup(metadata.ClearMetadataReportInstances)
+
+ // Build the MetadataInfo that the mock report will return
+ serviceURL, err := common.NewURL(
+ fmt.Sprintf("tri://127.0.0.1:20099/%s", testInterface),
+ common.WithInterface(testInterface),
+ common.WithMethods([]string{"Greet"}),
+ common.WithParamsValue(constant.ApplicationKey, testApp),
+ )
+ require.NoError(t, err)
+ svc := info.NewServiceInfoWithURL(serviceURL)
+ expectedMeta := info.NewMetadataInfoWithParams(testApp, revision,
map[string]*info.ServiceInfo{
+ svc.GetMatchKey(): svc,
+ })
+ mockReport.On("GetAppMetadata").Return(expectedMeta, nil).Once()
+
+ // Create a service instance that requests remote metadata storage
+ remoteInstance := ®istry.DefaultServiceInstance{
+ ID: "127.0.0.1:20099",
+ ServiceName: testApp,
+ Host: "127.0.0.1",
+ Port: 20099,
+ Enable: true,
+ Healthy: true,
+ Metadata: map[string]string{
+ constant.ExportedServicesRevisionPropertyName: revision,
+ constant.MetadataStorageTypePropertyName:
constant.RemoteMetadataStorageType,
+ constant.ServiceInstanceEndpoints:
`[{"port":20099,"protocol":"tri"}]`,
+ },
+ }
+
+ // Remove the cache entry after the test so it doesn't bleed into other
tests.
+ t.Cleanup(func() {
+ if metaCache != nil {
+ metaCache.Delete(listenerRegistryId + ":" + revision)
+ }
+ })
+
+ listener := NewServiceInstancesChangedListener(testApp,
listenerRegistryId, gxset.NewSet(testApp))
+ notify := &capturingNotifyListener{}
+ listener.AddListenerAndNotify(common.MatchKey(testInterface,
constant.TriProtocol), notify)
+
+ require.NoError(t,
listener.OnEvent(registry.NewServiceInstancesChangedEvent(testApp,
[]registry.ServiceInstance{remoteInstance})))
+
+ // The mock report must have been called exactly once, proving the
correct report was used
+ mockReport.AssertExpectations(t)
+ // The notify listener must have received the service from the fetched
metadata
+ assert.NotEmpty(t, notify.events, "expected service events from the
remote metadata fetch")
+}
+
+// listenerMockMetadataReport is a mock MetadataReport (and its factory) for
+// TestListenerUsesRegistryIdToFetchRemoteMetadata.
+type listenerMockMetadataReport struct {
+ mock.Mock
+}
+
+func (m *listenerMockMetadataReport) CreateMetadataReport(*common.URL)
metadatareport.MetadataReport {
+ return m
+}
+
+func (m *listenerMockMetadataReport) GetAppMetadata(string, string)
(*info.MetadataInfo, error) {
+ args := m.Called()
+ return args.Get(0).(*info.MetadataInfo), args.Error(1)
+}
+
+func (m *listenerMockMetadataReport) PublishAppMetadata(string, string,
*info.MetadataInfo) error {
+ args := m.Called()
+ return args.Error(0)
+}
+
+func (m *listenerMockMetadataReport) RegisterServiceAppMapping(string, string,
string) error {
+ args := m.Called()
+ return args.Error(0)
+}
+
+func (m *listenerMockMetadataReport) GetServiceAppMapping(string, string,
mapping.MappingListener) (*gxset.HashSet, error) {
+ args := m.Called()
+ return args.Get(0).(*gxset.HashSet), args.Error(1)
+}
+
+func (m *listenerMockMetadataReport) RemoveServiceAppMappingListener(string,
string) error {
+ args := m.Called()
+ return args.Error(0)
+}
+
func newTestServiceInstance(t *testing.T, port int, environment string)
registry.ServiceInstance {
t.Helper()
@@ -159,7 +268,9 @@ func newTestServiceInstance(t *testing.T, port int,
environment string) registry
func newTestServiceInstanceWithRevision(t *testing.T, port int, environment
string, revision string) registry.ServiceInstance {
t.Helper()
- metaCache.Set(revision, newTestMetadataInfo(t, revision, port,
environment))
+ // Pre-populate the cache under the composite key used by
GetMetadataInfo.
+ // All test listeners use constant.DefaultKey as their registryId.
+ metaCache.Set(testApp+":"+constant.DefaultKey+":"+revision,
newTestMetadataInfo(t, revision, port, environment))
return newTestServiceInstanceOnly(port, environment, revision)
}
