fix: [2.4] Check err is ErrKeyNotFound when CASCachedValue (#34489)

Cherry-pick from master
pr: #34488
See also #33785

When config item is not present in paramtable, CAS fails due to
GetConfig returns error.

This PR make this returned err instance of ErrKeyNotFound and check
error type in \`CASCachedValue\` methods.

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

pkg/config/config_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
 	"go.etcd.io/etcd/server/v3/embed"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/v3client"
@@ -30,7 +31,7 @@ import (
 func TestConfigFromEnv(t *testing.T) {
 	mgr, _ := Init()
 	_, err := mgr.GetConfig("test.env")
-	assert.EqualError(t, err, "key not found: test.env")
+	assert.ErrorIs(t, err, ErrKeyNotFound)
 
 	t.Setenv("TEST_ENV", "value")
 	mgr, _ = Init(WithEnvSource(formatKey))
@@ -67,7 +68,7 @@ func TestConfigFromRemote(t *testing.T) {
 
 	t.Run("origin is empty", func(t *testing.T) {
 		_, err = mgr.GetConfig("test.etcd")
-		assert.EqualError(t, err, "key not found: test.etcd")
+		assert.ErrorIs(t, err, ErrKeyNotFound)
 
 		client.KV.Put(ctx, "test/config/test/etcd", "value")
 
@@ -84,7 +85,7 @@ func TestConfigFromRemote(t *testing.T) {
 		time.Sleep(100 * time.Millisecond)
 
 		_, err = mgr.GetConfig("TEST_ETCD")
-		assert.EqualError(t, err, "key not found: TEST_ETCD")
+		assert.ErrorIs(t, err, ErrKeyNotFound)
 	})
 
 	t.Run("override origin value", func(t *testing.T) {
@@ -134,7 +135,7 @@ func TestConfigFromRemote(t *testing.T) {
 		client.KV.Put(ctx, "test/config/test/etcd", "value2")
 		assert.Eventually(t, func() bool {
 			_, err = mgr.GetConfig("test.etcd")
-			return err != nil && err.Error() == "key not found: test.etcd"
+			return err != nil && errors.Is(err, ErrKeyNotFound)
 		}, 300*time.Millisecond, 10*time.Millisecond)
 	})
 }

pkg/config/env_source.go

@@ -16,10 +16,10 @@
 package config
 
 import (
-	"fmt"
 	"os"
 	"strings"
 
+	"github.com/cockroachdb/errors"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
 )
 
@@ -51,7 +51,7 @@ func (es EnvSource) GetConfigurationByKey(key string) (string, error) {
 	value, ok := es.configs.Get(key)
 
 	if !ok {
-		return "", fmt.Errorf("key not found: %s", key)
+		return "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found: %s", key)
 	}
 
 	return value, nil

pkg/config/etcd_source.go

@@ -18,12 +18,12 @@ package config
 
 import (
 	"context"
-	"fmt"
 	"path"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/cockroachdb/errors"
 	"github.com/samber/lo"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/zap"
@@ -80,7 +80,7 @@ func (es *EtcdSource) GetConfigurationByKey(key string) (string, error) {
 	v, ok := es.currentConfigs[key]
 	es.RUnlock()
 	if !ok {
-		return "", fmt.Errorf("key not found: %s", key)
+		return "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found: %s", key)
 	}
 	return v, nil
 }

pkg/config/file_source.go

@@ -17,7 +17,6 @@
 package config
 
 import (
-	"fmt"
 	"os"
 	"sync"
 
@@ -55,7 +54,7 @@ func (fs *FileSource) GetConfigurationByKey(key string) (string, error) {
 	v, ok := fs.configs[key]
 	fs.RUnlock()
 	if !ok {
-		return "", fmt.Errorf("key not found: %s", key)
+		return "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found: %s", key)
 	}
 	return v, nil
 }

pkg/config/manager.go

@@ -119,7 +119,7 @@ func (m *Manager) CASCachedValue(key string, origin string, value interface{}) b
 	m.cacheMutex.Lock()
 	defer m.cacheMutex.Unlock()
 	current, err := m.GetConfig(key)
-	if err != nil {
+	if err != nil && !errors.Is(err, ErrKeyNotFound) {
 		return false
 	}
 	if current != origin {
@@ -149,13 +149,13 @@ func (m *Manager) GetConfig(key string) (string, error) {
 	v, ok := m.overlays.Get(realKey)
 	if ok {
 		if v == TombValue {
-			return "", fmt.Errorf("key not found %s", key)
+			return "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found %s", key)
 		}
 		return v, nil
 	}
 	sourceName, ok := m.keySourceMap.Get(realKey)
 	if !ok {
-		return "", fmt.Errorf("key not found: %s", key)
+		return "", errors.Wrap(ErrKeyNotFound, key) // fmt.Errorf("key not found: %s", key)
 	}
 	return m.getConfigValueBySource(realKey, sourceName)
 }

pkg/config/manager_test.go

@@ -239,7 +239,7 @@ func TestCachedConfig(t *testing.T) {
 		assert.False(t, exist)
 		mgr.CASCachedValue("cd", "", "xxx")
 		_, exist = mgr.GetCachedValue("cd")
-		assert.False(t, exist)
+		assert.True(t, exist)
 
 		// after refresh, the cached value should be reset
 		ctx := context.Background()