Repository: go-kratos/kratos
Pull request #3402 (2024-08-21 04:01:55)

fix: when entering idle mode, the old resolve-related resources have …
…not been completely processed, and an exception occurs when exiting idle mode immediately.

<!--
🎉 Thanks for sending a pull request to Kratos! Here are some tips for you:

1. If this is your first time contributing to Kratos, please read our contribution guide: https://go-kratos.dev/en/docs/community/contribution/
2. Ensure you have added or ran the appropriate tests and lint for your PR, please use `make lint` and `make test` before filing your PR, use `make clean` to tidy your go mod.
3. If the PR is unfinished, you may need to mark it as a WIP(Work In Progress) PR or Draft PR
4. Please use a semantic commits format title, such as `<type>[optional scope]: <description>`, see: https://go-kratos.dev/docs/community/contribution#type
5. at the same time, please note that similar work should be submitted in one PR as far as possible to reduce the workload of reviewers. Do not split a work into multiple PR unless it should.
-->

<!--
🎉 感谢您向 Kratos 发送 PR！以下是一些提示：
如果这是你第一次为 Kratos 贡献，请阅读我们的贡献指南：https://go-kratos.dev/en/docs/community/contribution/
2、确保您已经为您的 PR 添加或运行了适当的测试和lint，请在提交PR之前使用“make lint”和“make test”，使用“make clean”整理您的 go.mod。
3、如果 PR 未完成，您可能需要将其标记为 WIP（Work In Progress）PR 或 Draft PR
4、请使用语义提交格式标题，如“<类型>[可选范围]：<说明>`，请参阅：https://go-kratos.dev/docs/community/contribution#type
5. 同时请注意，同类的工作请尽量在一个PR中提交，以减轻 review 者的工作负担，不要把一项工作拆分成很多个PR，除非它应该这样做。
-->


#### Description (what this PR does / why we need it):
<!--
* The description should include the motivation for this PR or contrast this with previous behavior
-->
![image1](https://github.com/user-attachments/assets/0615ebc1-1c69-4467-9189-026eef4447f8)
![image2](https://github.com/user-attachments/assets/e296be12-b67a-470d-90c6-7f810dec9fe0)
 When grpc quickly enters and exits idle mode, there is a situation where it no longer obtains the latest service information. Previously, it was fixed that when the grpc client enters idle mode, the consul package will no longer watch service information again (https://github.com/go-kratos/kratos/pull/3162). 
 Recently, I found that it was not in idle mode. The target service was updated, but the client service discovery information was not updated. After investigation, it was found that there were problems with the previous modifications. A logic was added before. Every time it enters idle mode, if the number of watch objects in the serviceset reaches 0, it will try to recycle the serviceset. However, there is a certain delay in recycling the serviceset, and it may be blocked by a tick. Please see the code for details. When the serviceset has not been recycled, it will exit the idle mode immediately. When watching again, the resolve method will no longer be called. If the target service is no longer updated in the future, the client will no longer be able to obtain the latest service information because the goroutine for obtaining service information is no longer available.
What was changed?
1. Removed serviceset recycling
2. Added test case for this situation
3. No longer determine whether the serviceset exists before calling resolve.

#### Which issue(s) this PR fixes (resolves / be part of):
<!--
* Automatically closes linked issue when PR is merged.
* If your PR is not fully resolved the issue, please use `part of #<issue number>` instead.

Usage: `fixes/resolves #<issue number>`, or `fixes/resolves (paste link of issue)`.
-->


#### Other special notes for the reviewers:
<!--
* Somethings that need extra attention for the reviewers
* Some additional notes, TODO list, etc.
-->



Patch:
diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index e6ee0642..75a267ac 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -31,6 +31,6 @@ jobs:
       - name: Lint
         uses: golangci/golangci-lint-action@v6
         with:
-          version: v1.54.2
+          version: v1.60.3
           working-directory: ${{ matrix.workdir }}
           skip-pkg-cache: true
diff --git a/.golangci.yml b/.golangci.yml
index 556fcfbc..57b20dd7 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -17,7 +17,7 @@ linters:
     - gofumpt
     - goconst
     - goimports
-    - gomnd
+    - mnd
     - gocyclo
     - ineffassign
     - lll
@@ -54,9 +54,9 @@ linters-settings:
     multi-func: true
   lll:
     line-length: 160
-  gomnd:
+  mnd:
     # don't include the "operation", "argument" and "assign"
-    checks: 
+    checks:
       - case
       - condition
       - return
diff --git a/app.go b/app.go
index 1fa23132..86a6cb5b 100644
--- a/app.go
+++ b/app.go
@@ -9,12 +9,12 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/google/uuid"
+	"golang.org/x/sync/errgroup"
+
 	"github.com/go-kratos/kratos/v2/log"
 	"github.com/go-kratos/kratos/v2/registry"
 	"github.com/go-kratos/kratos/v2/transport"
-
-	"github.com/google/uuid"
-	"golang.org/x/sync/errgroup"
 )
 
 // AppInfo is application context value.
diff --git a/cmd/kratos/internal/proto/client/client.go b/cmd/kratos/internal/proto/client/client.go
index cb3f4a76..07e4b490 100644
--- a/cmd/kratos/internal/proto/client/client.go
+++ b/cmd/kratos/internal/proto/client/client.go
@@ -72,7 +72,7 @@ func walk(dir string, args []string) error {
 	if dir == "" {
 		dir = "."
 	}
-	return filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+	return filepath.Walk(dir, func(path string, _ os.FileInfo, _ error) error {
 		if ext := filepath.Ext(path); ext != ".proto" || strings.HasPrefix(path, "third_party") {
 			return nil
 		}
diff --git a/cmd/kratos/internal/run/run.go b/cmd/kratos/internal/run/run.go
index 2def97a1..3137016f 100644
--- a/cmd/kratos/internal/run/run.go
+++ b/cmd/kratos/internal/run/run.go
@@ -98,7 +98,7 @@ func findCMD(base string) (map[string]string, error) {
 	var root bool
 	next := func(dir string) (map[string]string, error) {
 		cmdPath := make(map[string]string)
-		err := filepath.Walk(dir, func(walkPath string, info os.FileInfo, err error) error {
+		err := filepath.Walk(dir, func(walkPath string, info os.FileInfo, _ error) error {
 			// multi level directory is not allowed under the cmdPath directory, so it is judged that the path ends with cmdPath.
 			if strings.HasSuffix(walkPath, "cmd") {
 				paths, err := os.ReadDir(walkPath)
diff --git a/cmd/protoc-gen-go-errors/buf.gen.yaml b/cmd/protoc-gen-go-errors/buf.gen.yaml
new file mode 100644
index 00000000..27ea54a0
--- /dev/null
+++ b/cmd/protoc-gen-go-errors/buf.gen.yaml
@@ -0,0 +1,6 @@
+version: v2
+plugins:
+  - remote: buf.build/protocolbuffers/go:v1.33.0
+    out: .
+    opt:
+      - paths=source_relative
diff --git a/cmd/protoc-gen-go-errors/buf.lock b/cmd/protoc-gen-go-errors/buf.lock
new file mode 100644
index 00000000..4f98143f
--- /dev/null
+++ b/cmd/protoc-gen-go-errors/buf.lock
@@ -0,0 +1,2 @@
+# Generated by buf. DO NOT EDIT.
+version: v2
diff --git a/cmd/protoc-gen-go-errors/buf.yaml b/cmd/protoc-gen-go-errors/buf.yaml
new file mode 100644
index 00000000..8c3d4ddb
--- /dev/null
+++ b/cmd/protoc-gen-go-errors/buf.yaml
@@ -0,0 +1,16 @@
+version: v2
+name: buf.build/go-kratos/protoc-gen-go-errors
+lint:
+  use:
+    - DEFAULT
+  except:
+    - PACKAGE_DIRECTORY_MATCH
+    - ENUM_VALUE_UPPER_SNAKE_CASE
+    - ENUM_VALUE_PREFIX
+    - ENUM_ZERO_VALUE_SUFFIX
+    - FIELD_LOWER_SNAKE_CASE
+deps:
+  - buf.build/googleapis/googleapis
+breaking:
+  use:
+    - FILE
diff --git a/config/env/env.go b/config/env/env.go
index cdd30e1b..ecd5c6ff 100644
--- a/config/env/env.go
+++ b/config/env/env.go
@@ -23,7 +23,7 @@ func (e *env) load(envs []string) []*config.KeyValue {
 	var kv []*config.KeyValue
 	for _, env := range envs {
 		var k, v string
-		subs := strings.SplitN(env, "=", 2) //nolint:gomnd
+		subs := strings.SplitN(env, "=", 2) //nolint:mnd
 		k = subs[0]
 		if len(subs) > 1 {
 			v = subs[1]
diff --git a/config/options.go b/config/options.go
index 95d6f6d3..7395660d 100644
--- a/config/options.go
+++ b/config/options.go
@@ -137,7 +137,7 @@ func resolver(input map[string]interface{}, mapper func(name string) string, toT
 
 func mapper(input map[string]interface{}) func(name string) string {
 	mapper := func(name string) string {
-		args := strings.SplitN(strings.TrimSpace(name), ":", 2) //nolint:gomnd
+		args := strings.SplitN(strings.TrimSpace(name), ":", 2) //nolint:mnd
 		if v, has := readValue(input, args[0]); has {
 			s, _ := v.String()
 			return s
@@ -183,7 +183,7 @@ func expand(s string, mapping func(string) string, toType bool) interface{} {
 	re := r.FindAllStringSubmatch(s, -1)
 	var ct interface{}
 	for _, i := range re {
-		if len(i) == 2 { //nolint:gomnd
+		if len(i) == 2 { //nolint:mnd
 			m := mapping(i[1])
 			if toType {
 				ct = convertToType(m)
diff --git a/contrib/log/zap/zap.go b/contrib/log/zap/zap.go
index acf453fe..812afb95 100644
--- a/contrib/log/zap/zap.go
+++ b/contrib/log/zap/zap.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 
 	"github.com/go-kratos/kratos/v2/log"
 )
@@ -32,6 +33,11 @@ func NewLogger(zlog *zap.Logger) *Logger {
 }
 
 func (l *Logger) Log(level log.Level, keyvals ...interface{}) error {
+	// If logging at this level is completely disabled, skip the overhead of
+	// string formatting.
+	if zapcore.Level(level) < zapcore.DPanicLevel && !l.log.Core().Enabled(zapcore.Level(level)) {
+		return nil
+	}
 	var (
 		msg    = ""
 		keylen = len(keyvals)
diff --git a/contrib/polaris/router.go b/contrib/polaris/router.go
index 2478b58b..93202eb5 100644
--- a/contrib/polaris/router.go
+++ b/contrib/polaris/router.go
@@ -139,7 +139,7 @@ func buildPolarisInstance(namespace string, nodes []selector.Node) *pb.ServiceIn
 		Service:   &v1.Service{Name: wrapperspb.String(nodes[0].ServiceName()), Namespace: wrapperspb.String("default")},
 		Instances: ins,
 	}
-	return pb.NewServiceInstancesInProto(d, func(s string) local.InstanceLocalValue {
+	return pb.NewServiceInstancesInProto(d, func(string) local.InstanceLocalValue {
 		return local.NewInstanceLocalValue()
 	}, &pb.SvcPluginValues{Routers: nil, Loadbalancer: nil}, nil)
 }
diff --git a/contrib/registry/consul/registry.go b/contrib/registry/consul/registry.go
index c83fa74c..5b098384 100644
--- a/contrib/registry/consul/registry.go
+++ b/contrib/registry/consul/registry.go
@@ -190,7 +190,6 @@ func (r *Registry) Watch(ctx context.Context, name string) (registry.Watcher, er
 			services:    &atomic.Value{},
 			serviceName: name,
 		}
-		set.ctx, set.cancel = context.WithCancel(context.Background())
 		r.registry[name] = set
 	}
 
@@ -209,10 +208,8 @@ func (r *Registry) Watch(ctx context.Context, name string) (registry.Watcher, er
 		// otherwise the initial data may be blocked forever during the watch.
 		w.event <- struct{}{}
 	}
-	if !ok {
-		if err := r.resolve(set.ctx, set); err != nil {
-			return nil, err
-		}
+	if err := r.resolve(ctx, set); err != nil {
+		return nil, err
 	}
 	return w, nil
 }
@@ -248,9 +245,6 @@ func (r *Registry) resolve(ctx context.Context, ss *serviceSet) error {
 				}
 				idx = tmpIdx
 			case <-ctx.Done():
-				r.lock.Lock()
-				delete(r.registry, ss.serviceName)
-				r.lock.Unlock()
 				return
 			}
 		}
diff --git a/contrib/registry/consul/service.go b/contrib/registry/consul/service.go
index 6567b772..a0a16609 100644
--- a/contrib/registry/consul/service.go
+++ b/contrib/registry/consul/service.go
@@ -1,7 +1,6 @@
 package consul
 
 import (
-	"context"
 	"sync"
 	"sync/atomic"
 
@@ -13,9 +12,6 @@ type serviceSet struct {
 	watcher     map[*watcher]struct{}
 	services    *atomic.Value
 	lock        sync.RWMutex
-
-	ctx    context.Context
-	cancel context.CancelFunc
 }
 
 func (s *serviceSet) broadcast(ss []*registry.ServiceInstance) {
diff --git a/contrib/registry/consul/watcher.go b/contrib/registry/consul/watcher.go
index f84caeb4..0a5d3575 100644
--- a/contrib/registry/consul/watcher.go
+++ b/contrib/registry/consul/watcher.go
@@ -36,9 +36,5 @@ func (w *watcher) Stop() error {
 	w.set.lock.Lock()
 	defer w.set.lock.Unlock()
 	delete(w.set.watcher, w)
-	// close resolve
-	if len(w.set.watcher) == 0 {
-		w.set.cancel()
-	}
 	return nil
 }
diff --git a/contrib/registry/etcd/registry.go b/contrib/registry/etcd/registry.go
index 476436e2..1a931153 100644
--- a/contrib/registry/etcd/registry.go
+++ b/contrib/registry/etcd/registry.go
@@ -162,7 +162,7 @@ func (r *Registry) heartBeat(ctx context.Context, leaseID clientv3.LeaseID, key
 	if err != nil {
 		curLeaseID = 0
 	}
-	rand.Seed(time.Now().Unix())
+	randSource := rand.New(rand.NewSource(time.Now().Unix()))
 
 	for {
 		if curLeaseID == 0 {
@@ -200,7 +200,7 @@ func (r *Registry) heartBeat(ctx context.Context, leaseID clientv3.LeaseID, key
 					break
 				}
 				retreat = append(retreat, 1<<retryCnt)
-				time.Sleep(time.Duration(retreat[rand.Intn(len(retreat))]) * time.Second)
+				time.Sleep(time.Duration(retreat[randSource.Intn(len(retreat))]) * time.Second)
 			}
 			if _, ok := <-kac; !ok {
 				// retry failed
diff --git a/contrib/registry/eureka/client.go b/contrib/registry/eureka/client.go
index 3b985c5b..0c19cd3f 100644
--- a/contrib/registry/eureka/client.go
+++ b/contrib/registry/eureka/client.go
@@ -292,10 +292,10 @@ func (e *Client) pickServer(currentTimes int) string {
 }
 
 func (e *Client) shuffle() {
-	rand.Seed(time.Now().UnixNano())
-	rand.Shuffle(len(e.urls), func(i, j int) {
-		e.urls[i], e.urls[j] = e.urls[j], e.urls[i]
-	})
+	rand.New(rand.NewSource(time.Now().UnixNano())).
+		Shuffle(len(e.urls), func(i, j int) {
+			e.urls[i], e.urls[j] = e.urls[j], e.urls[i]
+		})
 }
 
 func (e *Client) buildAPI(currentTimes int, params ...string) string {
diff --git a/contrib/registry/kubernetes/registry.go b/contrib/registry/kubernetes/registry.go
index 21562dc5..7379abf3 100644
--- a/contrib/registry/kubernetes/registry.go
+++ b/contrib/registry/kubernetes/registry.go
@@ -205,13 +205,13 @@ func (s *Registry) Watch(ctx context.Context, name string) (registry.Watcher, er
 			}
 		},
 		Handler: cache.ResourceEventHandlerFuncs{
-			AddFunc: func(obj interface{}) {
+			AddFunc: func(interface{}) {
 				s.sendLatestInstances(ctx, name, announcement)
 			},
-			UpdateFunc: func(oldObj, newObj interface{}) {
+			UpdateFunc: func(interface{}, interface{}) {
 				s.sendLatestInstances(ctx, name, announcement)
 			},
-			DeleteFunc: func(obj interface{}) {
+			DeleteFunc: func(interface{}) {
 				s.sendLatestInstances(ctx, name, announcement)
 			},
 		},
diff --git a/contrib/registry/nacos/watcher.go b/contrib/registry/nacos/watcher.go
index 7f1dc846..16cec4ab 100644
--- a/contrib/registry/nacos/watcher.go
+++ b/contrib/registry/nacos/watcher.go
@@ -40,7 +40,7 @@ func newWatcher(ctx context.Context, cli naming_client.INamingClient, serviceNam
 		ServiceName: serviceName,
 		Clusters:    clusters,
 		GroupName:   groupName,
-		SubscribeCallback: func(services []model.SubscribeService, err error) {
+		SubscribeCallback: func([]model.SubscribeService, error) {
 			select {
 			case w.watchChan <- struct{}{}:
 			default:
diff --git a/encoding/form/proto_decode.go b/encoding/form/proto_decode.go
index 5b68b92e..cdd93108 100644
--- a/encoding/form/proto_decode.go
+++ b/encoding/form/proto_decode.go
@@ -172,7 +172,7 @@ func parseField(fd protoreflect.FieldDescriptor, value string) (protoreflect.Val
 		}
 		v := enum.Descriptor().Values().ByName(protoreflect.Name(value))
 		if v == nil {
-			i, err := strconv.ParseInt(value, 10, 32) //nolint:gomnd
+			i, err := strconv.ParseInt(value, 10, 32) //nolint:mnd
 			if err != nil {
 				return protoreflect.Value{}, fmt.Errorf("%q is not a valid value", value)
 			}
@@ -183,37 +183,37 @@ func parseField(fd protoreflect.FieldDescriptor, value string) (protoreflect.Val
 		}
 		return protoreflect.ValueOfEnum(v.Number()), nil
 	case protoreflect.Int32Kind, protoreflect.Sint32Kind, protoreflect.Sfixed32Kind:
-		v, err := strconv.ParseInt(value, 10, 32) //nolint:gomnd
+		v, err := strconv.ParseInt(value, 10, 32) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		return protoreflect.ValueOfInt32(int32(v)), nil
 	case protoreflect.Int64Kind, protoreflect.Sint64Kind, protoreflect.Sfixed64Kind:
-		v, err := strconv.ParseInt(value, 10, 64) //nolint:gomnd
+		v, err := strconv.ParseInt(value, 10, 64) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		return protoreflect.ValueOfInt64(v), nil
 	case protoreflect.Uint32Kind, protoreflect.Fixed32Kind:
-		v, err := strconv.ParseUint(value, 10, 32) //nolint:gomnd
+		v, err := strconv.ParseUint(value, 10, 32) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		return protoreflect.ValueOfUint32(uint32(v)), nil
 	case protoreflect.Uint64Kind, protoreflect.Fixed64Kind:
-		v, err := strconv.ParseUint(value, 10, 64) //nolint:gomnd
+		v, err := strconv.ParseUint(value, 10, 64) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		return protoreflect.ValueOfUint64(v), nil
 	case protoreflect.FloatKind:
-		v, err := strconv.ParseFloat(value, 32) //nolint:gomnd
+		v, err := strconv.ParseFloat(value, 32) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		return protoreflect.ValueOfFloat32(float32(v)), nil
 	case protoreflect.DoubleKind:
-		v, err := strconv.ParseFloat(value, 64) //nolint:gomnd
+		v, err := strconv.ParseFloat(value, 64) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
@@ -255,37 +255,37 @@ func parseMessage(md protoreflect.MessageDescriptor, value string) (protoreflect
 		}
 		msg = durationpb.New(d)
 	case "google.protobuf.DoubleValue":
-		v, err := strconv.ParseFloat(value, 64) //nolint:gomnd
+		v, err := strconv.ParseFloat(value, 64) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		msg = wrapperspb.Double(v)
 	case "google.protobuf.FloatValue":
-		v, err := strconv.ParseFloat(value, 32) //nolint:gomnd
+		v, err := strconv.ParseFloat(value, 32) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		msg = wrapperspb.Float(float32(v))
 	case "google.protobuf.Int64Value":
-		v, err := strconv.ParseInt(value, 10, 64) //nolint:gomnd
+		v, err := strconv.ParseInt(value, 10, 64) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		msg = wrapperspb.Int64(v)
 	case "google.protobuf.Int32Value":
-		v, err := strconv.ParseInt(value, 10, 32) //nolint:gomnd
+		v, err := strconv.ParseInt(value, 10, 32) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		msg = wrapperspb.Int32(int32(v))
 	case "google.protobuf.UInt64Value":
-		v, err := strconv.ParseUint(value, 10, 64) //nolint:gomnd
+		v, err := strconv.ParseUint(value, 10, 64) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
 		msg = wrapperspb.UInt64(v)
 	case "google.protobuf.UInt32Value":
-		v, err := strconv.ParseUint(value, 10, 32) //nolint:gomnd
+		v, err := strconv.ParseUint(value, 10, 32) //nolint:mnd
 		if err != nil {
 			return protoreflect.Value{}, err
 		}
@@ -358,9 +358,9 @@ func parseURLQueryMapKey(key string) (string, string, error) {
 		endIndex   = strings.IndexByte(key, ']')
 	)
 	if startIndex < 0 {
-		//nolint:gomnd
+		//nolint:mnd
 		values := strings.Split(key, fieldSeparater)
-		//nolint:gomnd
+		//nolint:mnd
 		if len(values) != 2 {
 			return "", "", errInvalidFormatMapKey
 		}
diff --git a/errors/errors.go b/errors/errors.go
index caf31cd3..ae0764f8 100644
--- a/errors/errors.go
+++ b/errors/errors.go
@@ -89,7 +89,7 @@ func Errorf(code int, reason, format string, a ...interface{}) error {
 // It supports wrapped errors.
 func Code(err error) int {
 	if err == nil {
-		return 200 //nolint:gomnd
+		return 200 //nolint:mnd
 	}
 	return int(FromError(err).Code)
 }
diff --git a/errors/types.go b/errors/types.go
index 6a57f32a..fbd2cb1f 100644
--- a/errors/types.go
+++ b/errors/types.go
@@ -1,4 +1,4 @@
-// nolint:gomnd
+// nolint:mnd
 package errors
 
 // BadRequest new BadRequest error that is mapped to a 400 response.
diff --git a/internal/host/host.go b/internal/host/host.go
index 774f7ae3..5a6caba0 100644
--- a/internal/host/host.go
+++ b/internal/host/host.go
@@ -13,7 +13,7 @@ func ExtractHostPort(addr string) (host string, port uint64, err error) {
 	if err != nil {
 		return
 	}
-	port, err = strconv.ParseUint(ports, 10, 16) //nolint:gomnd
+	port, err = strconv.ParseUint(ports, 10, 16) //nolint:mnd
 	return
 }
 
diff --git a/middleware/metrics/metrics.go b/middleware/metrics/metrics.go
index 4459cdee..becc31fa 100644
--- a/middleware/metrics/metrics.go
+++ b/middleware/metrics/metrics.go
@@ -76,7 +76,7 @@ func DefaultSecondsHistogramView(histogramName string) metricsdk.View {
 					Boundaries: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.250, 0.5, 1},
 					NoMinMax:   true,
 				},
-				AttributeFilter: func(value attribute.KeyValue) bool {
+				AttributeFilter: func(attribute.KeyValue) bool {
 					return true
 				},
 			}, true
diff --git a/middleware/recovery/recovery.go b/middleware/recovery/recovery.go
index 53e67f81..4c02820e 100644
--- a/middleware/recovery/recovery.go
+++ b/middleware/recovery/recovery.go
@@ -36,7 +36,7 @@ func WithHandler(h HandlerFunc) Option {
 // Recovery is a server middleware that recovers from any panics.
 func Recovery(opts ...Option) middleware.Middleware {
 	op := options{
-		handler: func(ctx context.Context, req, err interface{}) error {
+		handler: func(context.Context, any, any) error {
 			return ErrUnknownRequest
 		},
 	}
@@ -48,7 +48,7 @@ func Recovery(opts ...Option) middleware.Middleware {
 			startTime := time.Now()
 			defer func() {
 				if rerr := recover(); rerr != nil {
-					buf := make([]byte, 64<<10) //nolint:gomnd
+					buf := make([]byte, 64<<10) //nolint:mnd
 					n := runtime.Stack(buf, false)
 					buf = buf[:n]
 					log.Context(ctx).Errorf("%v: %+v\n%s\n", rerr, req, buf)
diff --git a/middleware/tracing/span.go b/middleware/tracing/span.go
index a0a9d462..1fa241ec 100644
--- a/middleware/tracing/span.go
+++ b/middleware/tracing/span.go
@@ -104,7 +104,7 @@ func setServerSpan(ctx context.Context, span trace.Span, m interface{}) {
 func parseFullMethod(fullMethod string) (string, []attribute.KeyValue) {
 	name := strings.TrimLeft(fullMethod, "/")
 	parts := strings.SplitN(name, "/", 2)
-	if len(parts) != 2 { //nolint:gomnd
+	if len(parts) != 2 { //nolint:mnd
 		// Invalid format, does not follow `/package.service/method`.
 		return name, []attribute.KeyValue{attribute.Key("rpc.operation").String(fullMethod)}
 	}
diff --git a/selector/node/direct/direct.go b/selector/node/direct/direct.go
index d32fc314..eeb6e219 100644
--- a/selector/node/direct/direct.go
+++ b/selector/node/direct/direct.go
@@ -36,7 +36,7 @@ func (*Builder) Build(n selector.Node) selector.WeightedNode {
 func (n *Node) Pick() selector.DoneFunc {
 	now := time.Now().UnixNano()
 	atomic.StoreInt64(&n.lastPick, now)
-	return func(ctx context.Context, di selector.DoneInfo) {}
+	return func(context.Context, selector.DoneInfo) {}
 }
 
 // Weight is node effective weight
diff --git a/selector/node/ewma/node.go b/selector/node/ewma/node.go
index 540a0bb3..b4daeaa6 100644
--- a/selector/node/ewma/node.go
+++ b/selector/node/ewma/node.go
@@ -121,7 +121,7 @@ func (n *Node) Pick() selector.DoneFunc {
 	reqs := atomic.AddInt64(&n.reqs, 1)
 	slot := reqs % 200
 	swapped := atomic.CompareAndSwapInt64(&n.inflights[slot], 0, start)
-	return func(ctx context.Context, di selector.DoneInfo) {
+	return func(_ context.Context, di selector.DoneInfo) {
 		if swapped {
 			atomic.CompareAndSwapInt64(&n.inflights[slot], start, 0)
 		}
diff --git a/transport/grpc/client.go b/transport/grpc/client.go
index edc57d3e..5e34d4be 100644
--- a/transport/grpc/client.go
+++ b/transport/grpc/client.go
@@ -11,6 +11,7 @@ import (
 	grpcinsecure "google.golang.org/grpc/credentials/insecure"
 	grpcmd "google.golang.org/grpc/metadata"
 
+	"github.com/go-kratos/kratos/v2/internal/matcher"
 	"github.com/go-kratos/kratos/v2/log"
 	"github.com/go-kratos/kratos/v2/middleware"
 	"github.com/go-kratos/kratos/v2/registry"
@@ -114,8 +115,8 @@ func WithHealthCheck(healthCheck bool) ClientOption {
 
 // WithLogger with logger
 // Deprecated: use global logger instead.
-func WithLogger(_ log.Logger) ClientOption {
-	return func(o *clientOptions) {}
+func WithLogger(log.Logger) ClientOption {
+	return func(*clientOptions) {}
 }
 
 func WithPrintDiscoveryDebugLog(p bool) ClientOption {
@@ -132,6 +133,7 @@ type clientOptions struct {
 	timeout                time.Duration
 	discovery              registry.Discovery
 	middleware             []middleware.Middleware
+	streamMiddleware       []middleware.Middleware
 	ints                   []grpc.UnaryClientInterceptor
 	streamInts             []grpc.StreamClientInterceptor
 	grpcOpts               []grpc.DialOption
@@ -166,7 +168,7 @@ func dial(ctx context.Context, insecure bool, opts ...ClientOption) (*grpc.Clien
 		unaryClientInterceptor(options.middleware, options.timeout, options.filters),
 	}
 	sints := []grpc.StreamClientInterceptor{
-		streamClientInterceptor(options.filters),
+		streamClientInterceptor(options.streamMiddleware, options.filters),
 	}
 
 	if len(options.ints) > 0 {
@@ -239,7 +241,54 @@ func unaryClientInterceptor(ms []middleware.Middleware, timeout time.Duration, f
 	}
 }
 
-func streamClientInterceptor(filters []selector.NodeFilter) grpc.StreamClientInterceptor {
+// wrappedClientStream wraps the grpc.ClientStream and applies middleware
+type wrappedClientStream struct {
+	grpc.ClientStream
+	ctx        context.Context
+	middleware matcher.Matcher
+}
+
+func (w *wrappedClientStream) Context() context.Context {
+	return w.ctx
+}
+
+func (w *wrappedClientStream) SendMsg(m interface{}) error {
+	h := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return req, w.ClientStream.SendMsg(m)
+	}
+
+	info, ok := transport.FromClientContext(w.ctx)
+	if !ok {
+		return fmt.Errorf("transport value stored in ctx returns: %v", ok)
+	}
+
+	if next := w.middleware.Match(info.Operation()); len(next) > 0 {
+		h = middleware.Chain(next...)(h)
+	}
+
+	_, err := h(w.ctx, m)
+	return err
+}
+
+func (w *wrappedClientStream) RecvMsg(m interface{}) error {
+	h := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return req, w.ClientStream.RecvMsg(m)
+	}
+
+	info, ok := transport.FromClientContext(w.ctx)
+	if !ok {
+		return fmt.Errorf("transport value stored in ctx returns: %v", ok)
+	}
+
+	if next := w.middleware.Match(info.Operation()); len(next) > 0 {
+		h = middleware.Chain(next...)(h)
+	}
+
+	_, err := h(w.ctx, m)
+	return err
+}
+
+func streamClientInterceptor(ms []middleware.Middleware, filters []selector.NodeFilter) grpc.StreamClientInterceptor {
 	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) { // nolint
 		ctx = transport.NewClientContext(ctx, &Transport{
 			endpoint:    cc.Target(),
@@ -249,6 +298,28 @@ func streamClientInterceptor(filters []selector.NodeFilter) grpc.StreamClientInt
 		})
 		var p selector.Peer
 		ctx = selector.NewPeerContext(ctx, &p)
-		return streamer(ctx, desc, cc, method, opts...)
+
+		clientStream, err := streamer(ctx, desc, cc, method, opts...)
+		if err != nil {
+			return nil, err
+		}
+
+		h := func(ctx context.Context, req interface{}) (interface{}, error) {
+			return streamer, nil
+		}
+
+		m := matcher.New()
+		if len(ms) > 0 {
+			m.Use(ms...)
+			middleware.Chain(ms...)(h)
+		}
+
+		wrappedStream := &wrappedClientStream{
+			ClientStream: clientStream,
+			ctx:          ctx,
+			middleware:   m,
+		}
+
+		return wrappedStream, nil
 	}
 }
diff --git a/transport/grpc/interceptor.go b/transport/grpc/interceptor.go
index 6cc33154..6261442d 100644
--- a/transport/grpc/interceptor.go
+++ b/transport/grpc/interceptor.go
@@ -2,11 +2,13 @@ package grpc
 
 import (
 	"context"
+	"fmt"
 
 	"google.golang.org/grpc"
 	grpcmd "google.golang.org/grpc/metadata"
 
 	ic "github.com/go-kratos/kratos/v2/internal/context"
+	"github.com/go-kratos/kratos/v2/internal/matcher"
 	"github.com/go-kratos/kratos/v2/middleware"
 	"github.com/go-kratos/kratos/v2/transport"
 )
@@ -48,13 +50,15 @@ func (s *Server) unaryServerInterceptor() grpc.UnaryServerInterceptor {
 // wrappedStream is rewrite grpc stream's context
 type wrappedStream struct {
 	grpc.ServerStream
-	ctx context.Context
+	ctx        context.Context
+	middleware matcher.Matcher
 }
 
-func NewWrappedStream(ctx context.Context, stream grpc.ServerStream) grpc.ServerStream {
+func NewWrappedStream(ctx context.Context, stream grpc.ServerStream, m matcher.Matcher) grpc.ServerStream {
 	return &wrappedStream{
 		ServerStream: stream,
 		ctx:          ctx,
+		middleware:   m,
 	}
 }
 
@@ -76,7 +80,19 @@ func (s *Server) streamServerInterceptor() grpc.StreamServerInterceptor {
 			replyHeader: headerCarrier(replyHeader),
 		})
 
-		ws := NewWrappedStream(ctx, ss)
+		h := func(ctx context.Context, req interface{}) (interface{}, error) {
+			return handler(srv, ss), nil
+		}
+
+		if next := s.streamMiddleware.Match(info.FullMethod); len(next) > 0 {
+			middleware.Chain(next...)(h)
+		}
+
+		ctx = context.WithValue(ctx, stream{
+			ServerStream:     ss,
+			streamMiddleware: s.streamMiddleware,
+		}, ss)
+		ws := NewWrappedStream(ctx, ss, s.streamMiddleware)
 
 		err := handler(srv, ws)
 		if len(replyHeader) > 0 {
@@ -85,3 +101,48 @@ func (s *Server) streamServerInterceptor() grpc.StreamServerInterceptor {
 		return err
 	}
 }
+
+type stream struct {
+	grpc.ServerStream
+	streamMiddleware matcher.Matcher
+}
+
+func GetStream(ctx context.Context) grpc.ServerStream {
+	return ctx.Value(stream{}).(grpc.ServerStream)
+}
+
+func (w *wrappedStream) SendMsg(m interface{}) error {
+	h := func(_ context.Context, req interface{}) (interface{}, error) {
+		return req, w.ServerStream.SendMsg(m)
+	}
+
+	info, ok := transport.FromServerContext(w.ctx)
+	if !ok {
+		return fmt.Errorf("transport value stored in ctx returns: %v", ok)
+	}
+
+	if next := w.middleware.Match(info.Operation()); len(next) > 0 {
+		h = middleware.Chain(next...)(h)
+	}
+
+	_, err := h(w.ctx, m)
+	return err
+}
+
+func (w *wrappedStream) RecvMsg(m interface{}) error {
+	h := func(_ context.Context, req interface{}) (interface{}, error) {
+		return req, w.ServerStream.RecvMsg(m)
+	}
+
+	info, ok := transport.FromServerContext(w.ctx)
+	if !ok {
+		return fmt.Errorf("transport value stored in ctx returns: %v", ok)
+	}
+
+	if next := w.middleware.Match(info.Operation()); len(next) > 0 {
+		h = middleware.Chain(next...)(h)
+	}
+
+	_, err := h(w.ctx, m)
+	return err
+}
diff --git a/transport/grpc/server.go b/transport/grpc/server.go
index c8c3f4f3..2b260bf1 100644
--- a/transport/grpc/server.go
+++ b/transport/grpc/server.go
@@ -61,8 +61,8 @@ func Timeout(timeout time.Duration) ServerOption {
 
 // Logger with server logger.
 // Deprecated: use global logger instead.
-func Logger(_ log.Logger) ServerOption {
-	return func(s *Server) {}
+func Logger(log.Logger) ServerOption {
+	return func(*Server) {}
 }
 
 // Middleware with server middleware.
@@ -72,6 +72,12 @@ func Middleware(m ...middleware.Middleware) ServerOption {
 	}
 }
 
+func StreamMiddleware(m ...middleware.Middleware) ServerOption {
+	return func(s *Server) {
+		s.streamMiddleware.Use(m...)
+	}
+}
+
 // CustomHealth Checks server.
 func CustomHealth() ServerOption {
 	return func(s *Server) {
@@ -117,33 +123,35 @@ func Options(opts ...grpc.ServerOption) ServerOption {
 // Server is a gRPC server wrapper.
 type Server struct {
 	*grpc.Server
-	baseCtx      context.Context
-	tlsConf      *tls.Config
-	lis          net.Listener
-	err          error
-	network      string
-	address      string
-	endpoint     *url.URL
-	timeout      time.Duration
-	middleware   matcher.Matcher
-	unaryInts    []grpc.UnaryServerInterceptor
-	streamInts   []grpc.StreamServerInterceptor
-	grpcOpts     []grpc.ServerOption
-	health       *health.Server
-	customHealth bool
-	metadata     *apimd.Server
-	adminClean   func()
+	baseCtx          context.Context
+	tlsConf          *tls.Config
+	lis              net.Listener
+	err              error
+	network          string
+	address          string
+	endpoint         *url.URL
+	timeout          time.Duration
+	middleware       matcher.Matcher
+	streamMiddleware matcher.Matcher
+	unaryInts        []grpc.UnaryServerInterceptor
+	streamInts       []grpc.StreamServerInterceptor
+	grpcOpts         []grpc.ServerOption
+	health           *health.Server
+	customHealth     bool
+	metadata         *apimd.Server
+	adminClean       func()
 }
 
 // NewServer creates a gRPC server by options.
 func NewServer(opts ...ServerOption) *Server {
 	srv := &Server{
-		baseCtx:    context.Background(),
-		network:    "tcp",
-		address:    ":0",
-		timeout:    1 * time.Second,
-		health:     health.NewServer(),
-		middleware: matcher.New(),
+		baseCtx:          context.Background(),
+		network:          "tcp",
+		address:          ":0",
+		timeout:          1 * time.Second,
+		health:           health.NewServer(),
+		middleware:       matcher.New(),
+		streamMiddleware: matcher.New(),
 	}
 	for _, o := range opts {
 		o(srv)
diff --git a/transport/http/binding/encode.go b/transport/http/binding/encode.go
index 5f12cf37..2c1eda9c 100644
--- a/transport/http/binding/encode.go
+++ b/transport/http/binding/encode.go
@@ -20,7 +20,7 @@ func EncodeURL(pathTemplate string, msg interface{}, needQuery bool) string {
 	pathParams := make(map[string]struct{})
 	path := reg.ReplaceAllStringFunc(pathTemplate, func(in string) string {
 		// it's unreachable because the reg means that must have more than one char in {}
-		// if len(in) < 4 { //nolint:gomnd // **  explain the 4 number here :-) **
+		// if len(in) < 4 { //nolint:mnd // **  explain the 4 number here :-) **
 		//	return in
 		// }
 		key := in[1 : len(in)-1]
diff --git a/transport/http/client.go b/transport/http/client.go
index 1d74542d..d7efbcc8 100644
--- a/transport/http/client.go
+++ b/transport/http/client.go
@@ -252,7 +252,7 @@ func (client *Client) Invoke(ctx context.Context, method, path string, args inte
 }
 
 func (client *Client) invoke(ctx context.Context, req *http.Request, args interface{}, reply interface{}, c callInfo, opts ...CallOption) error {
-	h := func(ctx context.Context, in interface{}) (interface{}, error) {
+	h := func(ctx context.Context, _ interface{}) (interface{}, error) {
 		res, err := client.do(req.WithContext(ctx))
 		if res != nil {
 			cs := csAttempt{res: res}
diff --git a/transport/http/server.go b/transport/http/server.go
index 917b41e1..457fe724 100644
--- a/transport/http/server.go
+++ b/transport/http/server.go
@@ -58,8 +58,8 @@ func Timeout(timeout time.Duration) ServerOption {
 
 // Logger with server logger.
 // Deprecated: use global logger instead.
-func Logger(_ log.Logger) ServerOption {
-	return func(s *Server) {}
+func Logger(log.Logger) ServerOption {
+	return func(*Server) {}
 }
 
 // Middleware with service middleware option.
@@ -214,7 +214,7 @@ func (s *Server) Use(selector string, m ...middleware.Middleware) {
 
 // WalkRoute walks the router and all its sub-routers, calling walkFn for each route in the tree.
 func (s *Server) WalkRoute(fn WalkRouteFunc) error {
-	return s.router.Walk(func(route *mux.Route, router *mux.Router, ancestors []*mux.Route) error {
+	return s.router.Walk(func(route *mux.Route, _ *mux.Router, _ []*mux.Route) error {
 		methods, err := route.GetMethods()
 		if err != nil {
 			return nil // ignore no methods


Test patch:
diff --git a/config/config_test.go b/config/config_test.go
index ab234dd4..ec7fc9f7 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -154,8 +154,7 @@ func TestConfig(t *testing.T) {
 		t.Fatal("databaseDriver is not equal to val")
 	}
 
-	err = cf.Watch("endpoints", func(key string, value Value) {
-	})
+	err = cf.Watch("endpoints", func(string, Value) {})
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/config/options_test.go b/config/options_test.go
index 83f62b85..348471d3 100644
--- a/config/options_test.go
+++ b/config/options_test.go
@@ -387,7 +387,7 @@ func TestExpand(t *testing.T) {
 
 func TestWithMergeFunc(t *testing.T) {
 	c := &options{}
-	a := func(dst, src interface{}) error {
+	a := func(any, any) error {
 		return nil
 	}
 	WithMergeFunc(a)(c)
diff --git a/contrib/config/nacos/config_test.go b/contrib/config/nacos/config_test.go
index 2d290803..1667e99d 100644
--- a/contrib/config/nacos/config_test.go
+++ b/contrib/config/nacos/config_test.go
@@ -166,13 +166,13 @@ func TestConfig_Watch(t *testing.T) {
 				source: source,
 			},
 			wantErr: false,
-			processFunc: func(t *testing.T, w config.Watcher) {
+			processFunc: func(t *testing.T, _ config.Watcher) {
 				_, pErr := client.PublishConfig(vo.ConfigParam{DataId: "test.yaml", Group: "test", Content: "test: test"})
 				if pErr != nil {
 					t.Error(pErr)
 				}
 			},
-			deferFunc: func(t *testing.T, w config.Watcher) {
+			deferFunc: func(t *testing.T, _ config.Watcher) {
 				_, dErr := client.DeleteConfig(vo.ConfigParam{DataId: "test.yaml", Group: "test"})
 				if dErr != nil {
 					t.Error(dErr)
diff --git a/contrib/opensergo/opensergo_test.go b/contrib/opensergo/opensergo_test.go
index 0a2165fb..322c2c7e 100644
--- a/contrib/opensergo/opensergo_test.go
+++ b/contrib/opensergo/opensergo_test.go
@@ -309,7 +309,7 @@ func TestOpenSergo(t *testing.T) {
 			args: args{
 				opts: []Option{},
 			},
-			preFunc: func(t *testing.T) {
+			preFunc: func(_ *testing.T) {
 				err := os.Setenv("OPENSERGO_ENDPOINT", "127.0.0.1:9090")
 				if err != nil {
 					panic(err)
@@ -322,7 +322,7 @@ func TestOpenSergo(t *testing.T) {
 			args: args{
 				opts: []Option{},
 			},
-			preFunc: func(t *testing.T) {
+			preFunc: func(_ *testing.T) {
 				err := os.Setenv("OPENSERGO_BOOTSTRAP", `{"endpoint": "127.0.0.1:9090"}`)
 				if err != nil {
 					panic(err)
diff --git a/contrib/registry/consul/registry_test.go b/contrib/registry/consul/registry_test.go
index d6799b69..bb0aeef0 100644
--- a/contrib/registry/consul/registry_test.go
+++ b/contrib/registry/consul/registry_test.go
@@ -329,8 +329,7 @@ func TestRegistry_Watch(t *testing.T) {
 			},
 			want:    []*registry.ServiceInstance{instance1},
 			wantErr: false,
-			preFunc: func(t *testing.T) {
-			},
+			preFunc: func(*testing.T) {},
 		},
 		{
 			name: "ctx has been canceled",
@@ -344,8 +343,7 @@ func TestRegistry_Watch(t *testing.T) {
 			},
 			want:    nil,
 			wantErr: true,
-			preFunc: func(t *testing.T) {
-			},
+			preFunc: func(*testing.T) {},
 		},
 		{
 			name: "register with healthCheck",
@@ -432,10 +430,17 @@ func TestRegistry_IdleAndWatch(t *testing.T) {
 		Version:   "v0.0.1",
 		Endpoints: []string{fmt.Sprintf("tcp://%s?isSecure=false", addr)},
 	}
+	instance2 := &registry.ServiceInstance{
+		ID:        "1",
+		Name:      "server-1",
+		Version:   "v0.0.2",
+		Endpoints: []string{fmt.Sprintf("tcp://%s?isSecure=false", addr)},
+	}
 
 	type args struct {
-		ctx      context.Context
-		instance *registry.ServiceInstance
+		ctx            context.Context
+		instance       *registry.ServiceInstance
+		changeInstance *registry.ServiceInstance
 	}
 
 	tests := []struct {
@@ -447,8 +452,9 @@ func TestRegistry_IdleAndWatch(t *testing.T) {
 		{
 			name: "many client, one idle",
 			args: args{
-				ctx:      context.Background(),
-				instance: instance1,
+				ctx:            context.Background(),
+				instance:       instance1,
+				changeInstance: instance2,
 			},
 			want:    []*registry.ServiceInstance{instance1},
 			wantErr: false,
@@ -506,9 +512,7 @@ func TestRegistry_IdleAndWatch(t *testing.T) {
 				}
 			}
 			time.Sleep(2 * time.Second)
-			change := tt.args.instance
-			change.Version = "v0.0.2"
-			err = r.Register(tt.args.ctx, change)
+			err = r.Register(tt.args.ctx, tt.args.changeInstance)
 			if err != nil {
 				t.Error(err)
 			}
@@ -576,7 +580,7 @@ func TestRegistry_IdleAndWatch2(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.Background())
 			for i := 0; i < 10; i++ {
 				stopCtx, stopCancel := context.WithCancel(ctx)
-				watch, err1 := r.Watch(context.Background(), tt.args.instance.Name)
+				watch, err1 := r.Watch(stopCtx, tt.args.instance.Name)
 				if err1 != nil {
 					t.Error(err1)
 				}
@@ -588,10 +592,6 @@ func TestRegistry_IdleAndWatch2(t *testing.T) {
 						t.Errorf("GetService() got = %v", service)
 						return
 					}
-					_, err2 = watch.Next()
-					if err2 == nil {
-						t.Errorf("watch exit exception:%d ", i)
-					}
 				}(i)
 				go func() {
 					select {
@@ -633,6 +633,136 @@ func TestRegistry_IdleAndWatch2(t *testing.T) {
 	}
 }
 
+func TestRegistry_ExitOldResolverAndReWatch(t *testing.T) {
+	addr := fmt.Sprintf("%s:9091", getIntranetIP())
+
+	time.Sleep(time.Millisecond * 100)
+	cli, err := api.NewClient(&api.Config{Address: "127.0.0.1:8500", WaitTime: 2 * time.Second})
+	if err != nil {
+		t.Fatalf("create consul client failed: %v", err)
+	}
+
+	instance1 := &registry.ServiceInstance{
+		ID:        "1",
+		Name:      "server-1",
+		Version:   "v0.0.1",
+		Endpoints: []string{fmt.Sprintf("tcp://%s?isSecure=false", addr)},
+	}
+	instance2 := &registry.ServiceInstance{
+		ID:        "2",
+		Name:      "server-1",
+		Version:   "v0.0.2",
+		Endpoints: []string{fmt.Sprintf("tcp://%s?isSecure=false", addr)},
+	}
+	type args struct {
+		ctx             context.Context
+		opts            []Option
+		instance        *registry.ServiceInstance
+		initialInstance *registry.ServiceInstance
+	}
+
+	tests := []struct {
+		name    string
+		args    args
+		want    []*registry.ServiceInstance
+		wantErr bool
+	}{
+		{
+			name: "When it has entered idle mode, but the old resolver has not completely exited, the watch will be re-established due to new requests coming in.",
+			args: args{
+				ctx:             context.Background(),
+				initialInstance: instance1,
+				instance:        instance2,
+				opts: []Option{
+					WithHealthCheck(false),
+					WithTimeout(time.Second * 2),
+				},
+			},
+			want:    []*registry.ServiceInstance{instance2},
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := New(cli, tt.args.opts...)
+
+			err = r.Register(tt.args.ctx, tt.args.initialInstance)
+			if err != nil {
+				t.Error(err)
+			}
+			// first watch
+			ctx, cancel := context.WithCancel(context.Background())
+			watch, err := r.Watch(ctx, tt.args.instance.Name)
+			if err != nil {
+				t.Error(err)
+			}
+			service, err := watch.Next()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("GetService() error = %v, wantErr %v", err, tt.wantErr)
+				t.Errorf("GetService() got = %v", service)
+			}
+
+			time.Sleep(time.Second * 3)
+			// The simulation entered idle mode first, but the old resolver was not closed yet, and new requests triggered a new Watch.
+			watchCtx := context.Background()
+			// old resolver cancel
+			err = watch.Stop()
+			if err != nil {
+				t.Errorf("watch stop err:%v", err)
+			}
+			cancel()
+			// If it sleeps for a period of time, the old resolve goroutine will exit before the new Watch is processed, and there will be no problems at this time.
+			// time.Sleep(time.Second * 8)
+			newWatch, err := r.Watch(watchCtx, tt.args.instance.Name)
+			if err != nil {
+				t.Error(err)
+			}
+			service, err = newWatch.Next()
+			if (err != nil) != tt.wantErr {
+				t.Errorf("GetService() error = %v, wantErr %v", err, tt.wantErr)
+				t.Errorf("GetService() got = %v", service)
+			}
+			// change register info
+			time.Sleep(time.Second * 1)
+			err = r.Deregister(tt.args.ctx, tt.args.initialInstance)
+			if err != nil {
+				t.Error(err)
+			}
+			time.Sleep(time.Second * 5)
+			err = r.Register(tt.args.ctx, tt.args.instance)
+			if err != nil {
+				t.Error(err)
+			}
+
+			time.Sleep(time.Second * 2)
+
+			newWatchCtx, newWatchCancel := context.WithCancel(context.Background())
+			c := make(chan struct{}, 1)
+
+			go func() {
+				service, err = newWatch.Next()
+				if (err != nil) != tt.wantErr {
+					t.Errorf("GetService() error = %v, wantErr %v", err, tt.wantErr)
+					t.Errorf("GetService() got = %v", service)
+					return
+				}
+				if !reflect.DeepEqual(service, tt.want) {
+					t.Errorf("GetService() got = %v, want %v", service, tt.want)
+				}
+				c <- struct{}{}
+			}()
+			time.AfterFunc(time.Second*10, newWatchCancel)
+			select {
+			case <-newWatchCtx.Done():
+				t.Errorf("Timeout getservice. May be no new resolve goroutine to obtain the latest service information")
+			case <-c:
+				return
+			}
+		})
+	}
+}
+
 func getIntranetIP() string {
 	addrs, err := net.InterfaceAddrs()
 	if err != nil {
diff --git a/contrib/registry/kubernetes/registry_test.go b/contrib/registry/kubernetes/registry_test.go
index 2363b1c3..6f9761a4 100644
--- a/contrib/registry/kubernetes/registry_test.go
+++ b/contrib/registry/kubernetes/registry_test.go
@@ -110,7 +110,6 @@ func TestRegistry(t *testing.T) {
 		Endpoints: []string{"http://127.0.0.1:80"},
 	}
 	_, err = clientSet.AppsV1().Deployments(namespace).Create(context.Background(), &deployment, metav1.CreateOptions{})
-
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/contrib/registry/nacos/registry_test.go b/contrib/registry/nacos/registry_test.go
index 1910a68d..95768eea 100644
--- a/contrib/registry/nacos/registry_test.go
+++ b/contrib/registry/nacos/registry_test.go
@@ -530,7 +530,7 @@ func TestRegistry_Watch(t *testing.T) {
 			},
 			wantErr: true,
 			want:    nil,
-			processFunc: func(t *testing.T) {
+			processFunc: func(*testing.T) {
 				cancel()
 			},
 		},
diff --git a/contrib/registry/zookeeper/register_test.go b/contrib/registry/zookeeper/register_test.go
index a6f4515e..8ac9feec 100644
--- a/contrib/registry/zookeeper/register_test.go
+++ b/contrib/registry/zookeeper/register_test.go
@@ -80,7 +80,7 @@ func TestRegistry_GetService(t *testing.T) {
 		},
 		{
 			name: "conn close",
-			preFunc: func(t *testing.T) {
+			preFunc: func(*testing.T) {
 				conn.Close()
 			},
 			fields: fields{
@@ -339,7 +339,7 @@ func TestRegistry_Watch(t *testing.T) {
 					t.Error(err)
 				}
 			},
-			processFunc: func(t *testing.T, w registry.Watcher) {
+			processFunc: func(t *testing.T, _ registry.Watcher) {
 				err = r.Register(context.Background(), svrHello)
 				if err != nil {
 					t.Error(err)
@@ -357,7 +357,7 @@ func TestRegistry_Watch(t *testing.T) {
 			},
 			wantErr: true,
 			want:    nil,
-			processFunc: func(t *testing.T, w registry.Watcher) {
+			processFunc: func(*testing.T, registry.Watcher) {
 				cancel()
 			},
 		},
@@ -372,7 +372,7 @@ func TestRegistry_Watch(t *testing.T) {
 			},
 			wantErr: true,
 			want:    nil,
-			processFunc: func(t *testing.T, w registry.Watcher) {
+			processFunc: func(*testing.T, registry.Watcher) {
 				closeConn.Close()
 			},
 		},
diff --git a/encoding/encoding_test.go b/encoding/encoding_test.go
index 9746788a..98640ff4 100644
--- a/encoding/encoding_test.go
+++ b/encoding/encoding_test.go
@@ -2,7 +2,6 @@ package encoding
 
 import (
 	"encoding/xml"
-	"fmt"
 	"runtime/debug"
 	"testing"
 )
@@ -40,7 +39,7 @@ func TestRegisterCodec(t *testing.T) {
 	f := func() { RegisterCodec(nil) }
 	funcDidPanic, panicValue, _ := didPanic(f)
 	if !funcDidPanic {
-		t.Fatalf(fmt.Sprintf("func should panic\n\tPanic value:\t%#v", panicValue))
+		t.Fatalf("func should panic\n\tPanic value:\t%#v", panicValue)
 	}
 	if panicValue != "cannot register a nil Codec" {
 		t.Fatalf("panic error got %s want cannot register a nil Codec", panicValue)
@@ -50,7 +49,7 @@ func TestRegisterCodec(t *testing.T) {
 	}
 	funcDidPanic, panicValue, _ = didPanic(f)
 	if !funcDidPanic {
-		t.Fatalf(fmt.Sprintf("func should panic\n\tPanic value:\t%#v", panicValue))
+		t.Fatalf("func should panic\n\tPanic value:\t%#v", panicValue)
 	}
 	if panicValue != "cannot register Codec with empty string result for Name()" {
 		t.Fatalf("panic error got %s want cannot register Codec with empty string result for Name()", panicValue)
diff --git a/internal/matcher/middleware_test.go b/internal/matcher/middleware_test.go
index 9c948aeb..20575768 100644
--- a/internal/matcher/middleware_test.go
+++ b/internal/matcher/middleware_test.go
@@ -8,8 +8,8 @@ import (
 )
 
 func logging(module string) middleware.Middleware {
-	return func(handler middleware.Handler) middleware.Handler {
-		return func(ctx context.Context, req interface{}) (reply interface{}, err error) {
+	return func(middleware.Handler) middleware.Handler {
+		return func(context.Context, interface{}) (reply interface{}, err error) {
 			return module, nil
 		}
 	}
diff --git a/log/filter_test.go b/log/filter_test.go
index 8463a472..f95f4976 100644
--- a/log/filter_test.go
+++ b/log/filter_test.go
@@ -146,7 +146,10 @@ func testFilterFuncWithLoggerPrefix(level Level, keyvals ...interface{}) bool {
 }
 
 func TestFilterWithContext(t *testing.T) {
-	ctxKey := struct{}{}
+	type CtxKey struct {
+		Key string
+	}
+	ctxKey := CtxKey{Key: "context"}
 	ctxValue := "filter test value"
 
 	v1 := func() Valuer {
diff --git a/log/value_test.go b/log/value_test.go
index fd9167a6..defbbf74 100644
--- a/log/value_test.go
+++ b/log/value_test.go
@@ -19,7 +19,7 @@ func TestValue(t *testing.T) {
 	if got != v1 {
 		t.Errorf("Value() = %v, want %v", got, v1)
 	}
-	var v2 Valuer = func(ctx context.Context) interface{} {
+	var v2 Valuer = func(context.Context) interface{} {
 		return 3
 	}
 	got = Value(context.Background(), v2)
diff --git a/metadata/metadata_test.go b/metadata/metadata_test.go
index 9aa1c620..33f1b1a2 100644
--- a/metadata/metadata_test.go
+++ b/metadata/metadata_test.go
@@ -349,7 +349,7 @@ func TestMetadata_Range(t *testing.T) {
 		t.Errorf("metadata = %v, want %v", tmp, Metadata{"https://go-kratos.dev/": {"https://go-kratos.dev/"}, "kratos": {"kratos"}})
 	}
 	tmp = Metadata{}
-	md.Range(func(k string, v []string) bool {
+	md.Range(func(string, []string) bool {
 		return false
 	})
 	if !reflect.DeepEqual(tmp, Metadata{}) {
diff --git a/middleware/auth/jwt/jwt_test.go b/middleware/auth/jwt/jwt_test.go
index 5f2a3ba5..84a558e2 100644
--- a/middleware/auth/jwt/jwt_test.go
+++ b/middleware/auth/jwt/jwt_test.go
@@ -129,7 +129,7 @@ func TestJWTServerParse(t *testing.T) {
 		},
 	}
 
-	next := func(ctx context.Context, req interface{}) (interface{}, error) {
+	next := func(ctx context.Context, _ interface{}) (interface{}, error) {
 		testToken, _ := FromContext(ctx)
 		var name string
 		if customerClaims, ok := testToken.(*CustomerClaims); ok {
@@ -154,7 +154,7 @@ func TestJWTServerParse(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			server := Server(
-				func(token *jwt.Token) (interface{}, error) { return []byte(testKey), nil },
+				func(*jwt.Token) (interface{}, error) { return []byte(testKey), nil },
 				WithClaims(test.claims),
 			)(next)
 			wg := sync.WaitGroup{}
@@ -247,11 +247,11 @@ func TestServer(t *testing.T) {
 			}
 			var server middleware.Handler
 			if test.signingMethod != nil {
-				server = Server(func(token *jwt.Token) (interface{}, error) {
+				server = Server(func(*jwt.Token) (interface{}, error) {
 					return []byte(test.key), nil
 				}, WithSigningMethod(test.signingMethod))(next)
 			} else {
-				server = Server(func(token *jwt.Token) (interface{}, error) {
+				server = Server(func(*jwt.Token) (interface{}, error) {
 					return []byte(test.key), nil
 				})(next)
 			}
@@ -301,7 +301,7 @@ func TestClient(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			next := func(ctx context.Context, req interface{}) (interface{}, error) {
+			next := func(context.Context, any) (interface{}, error) {
 				return "reply", nil
 			}
 			handler := Client(test.tokenProvider)(next)
@@ -330,12 +330,12 @@ func TestTokenExpire(t *testing.T) {
 	}
 	token = fmt.Sprintf(bearerFormat, token)
 	time.Sleep(time.Second)
-	next := func(ctx context.Context, req interface{}) (interface{}, error) {
+	next := func(_ context.Context, req interface{}) (interface{}, error) {
 		t.Log(req)
 		return "reply", nil
 	}
 	ctx := transport.NewServerContext(context.Background(), &Transport{reqHeader: newTokenHeader(authorizationKey, token)})
-	server := Server(func(token *jwt.Token) (interface{}, error) {
+	server := Server(func(*jwt.Token) (interface{}, error) {
 		return []byte(testKey), nil
 	}, WithSigningMethod(jwt.SigningMethodHS256))(next)
 	_, err2 := server(ctx, "test expire token")
@@ -410,7 +410,7 @@ func TestClientWithClaims(t *testing.T) {
 	}
 
 	t.Run(test.name, func(t *testing.T) {
-		next := func(ctx context.Context, req interface{}) (interface{}, error) {
+		next := func(context.Context, interface{}) (interface{}, error) {
 			return "reply", nil
 		}
 		handler := Client(test.tokenProvider, WithClaims(mapClaimsFunc))(next)
@@ -446,7 +446,7 @@ func TestClientWithHeader(t *testing.T) {
 	tProvider := func(*jwt.Token) (interface{}, error) {
 		return []byte(testKey), nil
 	}
-	next := func(ctx context.Context, req interface{}) (interface{}, error) {
+	next := func(context.Context, interface{}) (interface{}, error) {
 		return "reply", nil
 	}
 	handler := Client(tProvider, WithClaims(mapClaimsFunc), WithTokenHeader(tokenHeader))(next)
@@ -484,7 +484,7 @@ func TestClientMissKey(t *testing.T) {
 	}
 
 	t.Run(test.name, func(t *testing.T) {
-		next := func(ctx context.Context, req interface{}) (interface{}, error) {
+		next := func(context.Context, interface{}) (interface{}, error) {
 			return "reply", nil
 		}
 		handler := Client(test.tokenProvider, WithClaims(mapClaimsFunc))(next)
diff --git a/middleware/circuitbreaker/circuitbreaker_test.go b/middleware/circuitbreaker/circuitbreaker_test.go
index 3b27da1a..3daca9f2 100644
--- a/middleware/circuitbreaker/circuitbreaker_test.go
+++ b/middleware/circuitbreaker/circuitbreaker_test.go
@@ -58,10 +58,10 @@ func Test_WithGroup(t *testing.T) {
 }
 
 func TestServer(_ *testing.T) {
-	nextValid := func(ctx context.Context, req interface{}) (interface{}, error) {
+	nextValid := func(context.Context, interface{}) (interface{}, error) {
 		return "Hello valid", nil
 	}
-	nextInvalid := func(ctx context.Context, req interface{}) (interface{}, error) {
+	nextInvalid := func(context.Context, interface{}) (interface{}, error) {
 		return nil, kratoserrors.InternalServer("", "")
 	}
 
diff --git a/middleware/logging/logging_test.go b/middleware/logging/logging_test.go
index 1e949518..87e3959a 100644
--- a/middleware/logging/logging_test.go
+++ b/middleware/logging/logging_test.go
@@ -87,7 +87,7 @@ func TestHTTP(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			bf.Reset()
-			next := func(ctx context.Context, req interface{}) (interface{}, error) {
+			next := func(context.Context, interface{}) (interface{}, error) {
 				return "reply", test.err
 			}
 			next = test.kind(logger)(next)
diff --git a/middleware/metrics/metrics_test.go b/middleware/metrics/metrics_test.go
index 3f30ac1a..348fdaee 100644
--- a/middleware/metrics/metrics_test.go
+++ b/middleware/metrics/metrics_test.go
@@ -77,7 +77,7 @@ func init() {
 				Name:        instrument.Name,
 				Description: instrument.Description,
 				Unit:        instrument.Unit,
-				AttributeFilter: func(value attribute.KeyValue) bool {
+				AttributeFilter: func(attribute.KeyValue) bool {
 					return true
 				},
 			}, true
@@ -139,10 +139,10 @@ func TestServer(t *testing.T) {
 	defer span.End()
 
 	e := errors.New("got an error")
-	nextError := func(ctx context.Context, req interface{}) (interface{}, error) {
+	nextError := func(context.Context, interface{}) (interface{}, error) {
 		return nil, e
 	}
-	nextValid := func(ctx context.Context, req interface{}) (interface{}, error) {
+	nextValid := func(context.Context, interface{}) (interface{}, error) {
 		time.Sleep(time.Millisecond * time.Duration(rand.Int31n(100)))
 		return "Hello valid", nil
 	}
@@ -201,10 +201,10 @@ func TestClient(t *testing.T) {
 	defer span.End()
 
 	e := errors.New("got an error")
-	nextError := func(ctx context.Context, req interface{}) (interface{}, error) {
+	nextError := func(context.Context, interface{}) (interface{}, error) {
 		return nil, e
 	}
-	nextValid := func(ctx context.Context, req interface{}) (interface{}, error) {
+	nextValid := func(context.Context, interface{}) (interface{}, error) {
 		return "Hello valid", nil
 	}
 
diff --git a/middleware/middleware_test.go b/middleware/middleware_test.go
index 11986b0a..4714ff97 100644
--- a/middleware/middleware_test.go
+++ b/middleware/middleware_test.go
@@ -10,8 +10,10 @@ import (
 var i int
 
 func TestChain(t *testing.T) {
-	next := func(ctx context.Context, req interface{}) (interface{}, error) {
-		t.Log(req)
+	next := func(_ context.Context, req interface{}) (interface{}, error) {
+		if req != "hello kratos!" {
+			t.Errorf("expect %v, got %v", "hello kratos!", req)
+		}
 		i += 10
 		return "reply", nil
 	}
diff --git a/middleware/ratelimit/ratelimit_test.go b/middleware/ratelimit/ratelimit_test.go
index b1422347..20cf9934 100644
--- a/middleware/ratelimit/ratelimit_test.go
+++ b/middleware/ratelimit/ratelimit_test.go
@@ -41,7 +41,7 @@ func Test_WithLimiter(t *testing.T) {
 }
 
 func TestServer(t *testing.T) {
-	nextValid := func(ctx context.Context, req interface{}) (interface{}, error) {
+	nextValid := func(context.Context, interface{}) (interface{}, error) {
 		return "Hello valid", nil
 	}
 
diff --git a/middleware/recovery/recovery_test.go b/middleware/recovery/recovery_test.go
index b20b5652..915861a2 100644
--- a/middleware/recovery/recovery_test.go
+++ b/middleware/recovery/recovery_test.go
@@ -15,7 +15,7 @@ func TestOnce(t *testing.T) {
 		}
 	}()
 
-	next := func(ctx context.Context, req interface{}) (interface{}, error) {
+	next := func(context.Context, interface{}) (interface{}, error) {
 		panic("panic reason")
 	}
 	_, e := Recovery(WithHandler(func(ctx context.Context, _, err interface{}) error {
@@ -29,11 +29,11 @@ func TestOnce(t *testing.T) {
 }
 
 func TestNotPanic(t *testing.T) {
-	next := func(ctx context.Context, req interface{}) (interface{}, error) {
+	next := func(_ context.Context, req interface{}) (interface{}, error) {
 		return req.(string) + "https://go-kratos.dev", nil
 	}
 
-	_, e := Recovery(WithHandler(func(ctx context.Context, req, err interface{}) error {
+	_, e := Recovery(WithHandler(func(_ context.Context, _ any, err any) error {
 		return errors.InternalServer("RECOVERY", fmt.Sprintf("panic triggered: %v", err))
 	}))(next)(context.Background(), "notPanic")
 	if e != nil {
diff --git a/middleware/selector/selector_test.go b/middleware/selector/selector_test.go
index 0835ef84..1769f7c0 100644
--- a/middleware/selector/selector_test.go
+++ b/middleware/selector/selector_test.go
@@ -96,7 +96,7 @@ func TestMatch(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			next := func(ctx context.Context, req interface{}) (interface{}, error) {
+			next := func(_ context.Context, req interface{}) (interface{}, error) {
 				t.Log(req)
 				return "reply", nil
 			}
@@ -132,7 +132,7 @@ func TestMatchClient(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			next := func(ctx context.Context, req interface{}) (interface{}, error) {
+			next := func(_ context.Context, req interface{}) (interface{}, error) {
 				t.Log(req)
 				return "reply", nil
 			}
@@ -167,11 +167,11 @@ func TestFunc(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			next := func(ctx context.Context, req interface{}) (interface{}, error) {
+			next := func(_ context.Context, req interface{}) (interface{}, error) {
 				t.Log(req)
 				return "reply", nil
 			}
-			next = Server(testMiddleware).Match(func(ctx context.Context, operation string) bool {
+			next = Server(testMiddleware).Match(func(_ context.Context, operation string) bool {
 				if strings.HasPrefix(operation, "/go-kratos.dev") || strings.HasSuffix(operation, "world") {
 					return true
 				}
@@ -224,11 +224,11 @@ func TestHeaderFunc(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			next := func(ctx context.Context, req interface{}) (interface{}, error) {
+			next := func(_ context.Context, req interface{}) (interface{}, error) {
 				t.Log(req)
 				return "reply", nil
 			}
-			next = Server(testMiddleware).Match(func(ctx context.Context, operation string) bool {
+			next = Server(testMiddleware).Match(func(ctx context.Context, _ string) bool {
 				tr, ok := transport.FromServerContext(ctx)
 				if !ok {
 					return false
diff --git a/middleware/tracing/tracing_test.go b/middleware/tracing/tracing_test.go
index 72740baf..fe902802 100644
--- a/middleware/tracing/tracing_test.go
+++ b/middleware/tracing/tracing_test.go
@@ -166,7 +166,6 @@ func TestServer(t *testing.T) {
 		WithTracerProvider(tracesdk.NewTracerProvider()),
 		WithPropagator(propagation.NewCompositeTextMapPropagator(propagation.Baggage{}, propagation.TraceContext{})),
 	)(next)(context.Background(), "test server: ")
-
 	if err != nil {
 		t.Errorf("expected error, got nil")
 	}
diff --git a/middleware/validate/validate_test.go b/middleware/validate/validate_test.go
index 34fc3349..34bea301 100644
--- a/middleware/validate/validate_test.go
+++ b/middleware/validate/validate_test.go
@@ -24,7 +24,7 @@ func (v protoVali) Validate() error {
 }
 
 func TestTable(t *testing.T) {
-	var mock middleware.Handler = func(ctx context.Context, req interface{}) (interface{}, error) { return nil, nil }
+	var mock middleware.Handler = func(context.Context, interface{}) (interface{}, error) { return nil, nil }
 
 	tests := []protoVali{
 		{"v1", 365, false},
diff --git a/options_test.go b/options_test.go
index e1b7288b..03cf4f51 100644
--- a/options_test.go
+++ b/options_test.go
@@ -66,9 +66,11 @@ func TestEndpoint(t *testing.T) {
 }
 
 func TestContext(t *testing.T) {
-	type ctxKey = struct{}
+	type ctxKey struct {
+		Key string
+	}
 	o := &options{}
-	v := context.WithValue(context.TODO(), ctxKey{}, "b")
+	v := context.WithValue(context.TODO(), ctxKey{Key: "context"}, "b")
 	Context(v)(o)
 	if !reflect.DeepEqual(v, o.ctx) {
 		t.Fatalf("o.ctx:%s is not equal to v:%s", o.ctx, v)
diff --git a/selector/selector_test.go b/selector/selector_test.go
index b7b963f0..75d3887a 100644
--- a/selector/selector_test.go
+++ b/selector/selector_test.go
@@ -37,7 +37,7 @@ func (n *mockWeightedNode) Weight() float64 {
 func (n *mockWeightedNode) Pick() DoneFunc {
 	now := time.Now().UnixNano()
 	atomic.StoreInt64(&n.lastPick, now)
-	return func(ctx context.Context, di DoneInfo) {}
+	return func(context.Context, DoneInfo) {}
 }
 
 // PickElapsed is time elapsed since the latest pick
diff --git a/transport/grpc/client_test.go b/transport/grpc/client_test.go
index 2d88f95a..cdbee600 100644
--- a/transport/grpc/client_test.go
+++ b/transport/grpc/client_test.go
@@ -84,7 +84,7 @@ func TestUnaryClientInterceptor(t *testing.T) {
 	resp := &struct{}{}
 
 	err := f(context.TODO(), "hello", req, resp, &grpc.ClientConn{},
-		func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, opts ...grpc.CallOption) error {
+		func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error {
 			return nil
 		})
 	if err != nil {
@@ -95,10 +95,10 @@ func TestUnaryClientInterceptor(t *testing.T) {
 func TestWithUnaryInterceptor(t *testing.T) {
 	o := &clientOptions{}
 	v := []grpc.UnaryClientInterceptor{
-		func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		func(context.Context, string, any, any, *grpc.ClientConn, grpc.UnaryInvoker, ...grpc.CallOption) error {
 			return nil
 		},
-		func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		func(context.Context, string, any, any, *grpc.ClientConn, grpc.UnaryInvoker, ...grpc.CallOption) error {
 			return nil
 		},
 	}
diff --git a/transport/grpc/server_test.go b/transport/grpc/server_test.go
index d9d2c7af..0117db9a 100644
--- a/transport/grpc/server_test.go
+++ b/transport/grpc/server_test.go
@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 
 	"github.com/go-kratos/kratos/v2/errors"
 	"github.com/go-kratos/kratos/v2/internal/matcher"
@@ -83,7 +84,7 @@ func TestServer(t *testing.T) {
 					return handler(ctx, req)
 				}
 			}),
-		UnaryInterceptor(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+		UnaryInterceptor(func(ctx context.Context, req interface{}, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
 			return handler(ctx, req)
 		}),
 		Options(grpc.InitialConnWindowSize(0)),
@@ -212,10 +213,10 @@ func TestTLSConfig(t *testing.T) {
 func TestUnaryInterceptor(t *testing.T) {
 	o := &Server{}
 	v := []grpc.UnaryServerInterceptor{
-		func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+		func(context.Context, interface{}, *grpc.UnaryServerInfo, grpc.UnaryHandler) (resp interface{}, err error) {
 			return nil, nil
 		},
-		func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+		func(context.Context, interface{}, *grpc.UnaryServerInfo, grpc.UnaryHandler) (resp interface{}, err error) {
 			return nil, nil
 		},
 	}
@@ -228,10 +229,10 @@ func TestUnaryInterceptor(t *testing.T) {
 func TestStreamInterceptor(t *testing.T) {
 	o := &Server{}
 	v := []grpc.StreamServerInterceptor{
-		func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		func(interface{}, grpc.ServerStream, *grpc.StreamServerInfo, grpc.StreamHandler) error {
 			return nil
 		},
-		func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		func(interface{}, grpc.ServerStream, *grpc.StreamServerInfo, grpc.StreamHandler) error {
 			return nil
 		},
 	}
@@ -269,7 +270,7 @@ func TestServer_unaryServerInterceptor(t *testing.T) {
 	}
 	srv.middleware.Use(EmptyMiddleware())
 	req := &struct{}{}
-	rv, err := srv.unaryServerInterceptor()(context.TODO(), req, &grpc.UnaryServerInfo{}, func(ctx context.Context, req interface{}) (i interface{}, e error) {
+	rv, err := srv.unaryServerInterceptor()(context.TODO(), req, &grpc.UnaryServerInfo{}, func(context.Context, interface{}) (interface{}, error) {
 		return &testResp{Data: "hi"}, nil
 	})
 	if err != nil {
@@ -280,6 +281,82 @@ func TestServer_unaryServerInterceptor(t *testing.T) {
 	}
 }
 
+type mockServerStream struct {
+	ctx      context.Context
+	sentMsg  interface{}
+	recvMsg  interface{}
+	metadata metadata.MD
+	grpc.ServerStream
+}
+
+func (m *mockServerStream) SetHeader(md metadata.MD) error {
+	m.metadata = md
+	return nil
+}
+
+func (m *mockServerStream) SendHeader(md metadata.MD) error {
+	m.metadata = md
+	return nil
+}
+
+func (m *mockServerStream) SetTrailer(md metadata.MD) {
+	m.metadata = md
+}
+
+func (m *mockServerStream) Context() context.Context {
+	return m.ctx
+}
+
+func (m *mockServerStream) SendMsg(msg interface{}) error {
+	m.sentMsg = msg
+	return nil
+}
+
+func (m *mockServerStream) RecvMsg(msg interface{}) error {
+	m.recvMsg = msg
+	return nil
+}
+
+func TestServer_streamServerInterceptor(t *testing.T) {
+	u, err := url.Parse("grpc://hello/world")
+	if err != nil {
+		t.Errorf("expect %v, got %v", nil, err)
+	}
+	srv := &Server{
+		baseCtx:          context.Background(),
+		endpoint:         u,
+		timeout:          time.Duration(10),
+		middleware:       matcher.New(),
+		streamMiddleware: matcher.New(),
+	}
+
+	srv.streamMiddleware.Use(EmptyMiddleware())
+
+	mockStream := &mockServerStream{
+		ctx: srv.baseCtx,
+	}
+
+	handler := func(_ interface{}, stream grpc.ServerStream) error {
+		resp := &testResp{Data: "stream hi"}
+		return stream.SendMsg(resp)
+	}
+
+	info := &grpc.StreamServerInfo{
+		FullMethod: "/grpc.reflection.v1.ServerReflection/ServerReflectionInfo",
+	}
+
+	err = srv.streamServerInterceptor()(nil, mockStream, info, handler)
+	if err != nil {
+		t.Errorf("expect %v, got %v", nil, err)
+	}
+
+	// Check response
+	resp := mockStream.sentMsg.(*testResp)
+	if !reflect.DeepEqual("stream hi", resp.Data) {
+		t.Errorf("expect %s, got %s", "stream hi", resp.Data)
+	}
+}
+
 func TestListener(t *testing.T) {
 	lis, err := net.Listen("tcp", ":0")
 	if err != nil {
diff --git a/transport/http/client_test.go b/transport/http/client_test.go
index 25ae9a8e..13bf4428 100644
--- a/transport/http/client_test.go
+++ b/transport/http/client_test.go
@@ -42,6 +42,15 @@ func (x *mockCallOption) after(_ *callInfo, _ *csAttempt) {
 	log.Println("run in mockCallOption.after")
 }
 
+func TestWithSubset(t *testing.T) {
+	co := &clientOptions{}
+	o := WithSubset(1)
+	o(co)
+	if co.subsetSize != 1 {
+		t.Error("expected subset size to be 1")
+	}
+}
+
 func TestWithTransport(t *testing.T) {
 	ov := &mockRoundTripper{}
 	o := WithTransport(ov)
@@ -71,10 +80,6 @@ func TestWithBlock(t *testing.T) {
 	}
 }
 
-func TestWithBalancer(_ *testing.T) {
-	// TODO
-}
-
 func TestWithTLSConfig(t *testing.T) {
 	ov := &tls.Config{}
 	o := WithTLSConfig(ov)
@@ -118,7 +123,7 @@ func TestWithEndpoint(t *testing.T) {
 
 func TestWithRequestEncoder(t *testing.T) {
 	o := &clientOptions{}
-	v := func(ctx context.Context, contentType string, in interface{}) (body []byte, err error) {
+	v := func(context.Context, string, interface{}) (body []byte, err error) {
 		return nil, nil
 	}
 	WithRequestEncoder(v)(o)
@@ -129,7 +134,7 @@ func TestWithRequestEncoder(t *testing.T) {
 
 func TestWithResponseDecoder(t *testing.T) {
 	o := &clientOptions{}
-	v := func(ctx context.Context, res *http.Response, out interface{}) error { return nil }
+	v := func(context.Context, *http.Response, interface{}) error { return nil }
 	WithResponseDecoder(v)(o)
 	if o.decoder == nil {
 		t.Errorf("expected encoder to be not nil")
@@ -138,7 +143,7 @@ func TestWithResponseDecoder(t *testing.T) {
 
 func TestWithErrorDecoder(t *testing.T) {
 	o := &clientOptions{}
-	v := func(ctx context.Context, res *http.Response) error { return nil }
+	v := func(context.Context, *http.Response) error { return nil }
 	WithErrorDecoder(v)(o)
 	if o.errorDecoder == nil {
 		t.Errorf("expected encoder to be not nil")
@@ -354,7 +359,7 @@ func TestNewClient(t *testing.T) {
 	if err == nil {
 		t.Error("err should be equal to callOption err")
 	}
-	client.opts.encoder = func(ctx context.Context, contentType string, in interface{}) (body []byte, err error) {
+	client.opts.encoder = func(context.Context, string, interface{}) (body []byte, err error) {
 		return nil, errors.New("mock test encoder error")
 	}
 	err = client.Invoke(context.Background(), http.MethodPost, "/go", map[string]string{"name": "kratos"}, nil, EmptyCallOption{})
diff --git a/transport/http/context_test.go b/transport/http/context_test.go
index 8cc79da0..b4949a02 100644
--- a/transport/http/context_test.go
+++ b/transport/http/context_test.go
@@ -102,7 +102,7 @@ func TestContextResponse(t *testing.T) {
 
 func TestResponseUnwrap(t *testing.T) {
 	res := httptest.NewRecorder()
-	f := func(rw http.ResponseWriter, r *http.Request, a interface{}) error {
+	f := func(rw http.ResponseWriter, _ *http.Request, _ interface{}) error {
 		u, ok := rw.(interface {
 			Unwrap() http.ResponseWriter
 		})
diff --git a/transport/http/router_test.go b/transport/http/router_test.go
index e6def62e..00b1a7dd 100644
--- a/transport/http/router_test.go
+++ b/transport/http/router_test.go
@@ -75,7 +75,7 @@ func TestRoute(t *testing.T) {
 		if err := ctx.Bind(u); err != nil {
 			return err
 		}
-		h := ctx.Middleware(func(ctx context.Context, in interface{}) (interface{}, error) {
+		h := ctx.Middleware(func(context.Context, interface{}) (interface{}, error) {
 			return u, nil
 		})
 		return ctx.Returns(h(ctx, u))
@@ -184,7 +184,7 @@ func TestRouter_Group(t *testing.T) {
 
 func TestHandle(_ *testing.T) {
 	r := newRouter("/", NewServer())
-	h := func(i Context) error {
+	h := func(Context) error {
 		return nil
 	}
 	r.GET("/get", h)
diff --git a/transport/http/server_test.go b/transport/http/server_test.go
index 44477504..c692a4ff 100644
--- a/transport/http/server_test.go
+++ b/transport/http/server_test.go
@@ -48,7 +48,7 @@ func TestServeHTTP(t *testing.T) {
 	}
 	mux := NewServer(Listener(ln))
 	mux.HandleFunc("/index", h)
-	mux.Route("/errors").GET("/cause", func(ctx Context) error {
+	mux.Route("/errors").GET("/cause", func(Context) error {
 		return kratoserrors.BadRequest("xxx", "zzz").
 			WithMetadata(map[string]string{"foo": "bar"}).
 			WithCause(errors.New("error cause"))
@@ -86,7 +86,7 @@ func TestServer(t *testing.T) {
 	srv.HandleHeader("content-type", "application/grpc-web+json", func(w http.ResponseWriter, r *http.Request) {
 		_ = json.NewEncoder(w).Encode(testData{Path: r.RequestURI})
 	})
-	srv.Route("/errors").GET("/cause", func(ctx Context) error {
+	srv.Route("/errors").GET("/cause", func(Context) error {
 		return kratoserrors.BadRequest("xxx", "zzz").
 			WithMetadata(map[string]string{"foo": "bar"}).
 			WithCause(errors.New("error cause"))
@@ -153,8 +153,7 @@ func testHeader(t *testing.T, srv *Server) {
 	if err != nil {
 		t.Errorf("expected nil got %v", err)
 	}
-	reqURL := fmt.Sprintf(e.String() + "/index")
-	req, err := http.NewRequest(http.MethodGet, reqURL, nil)
+	req, err := http.NewRequest(http.MethodGet, e.String()+"/index", nil)
 	if err != nil {
 		t.Errorf("expected nil got %v", err)
 	}
@@ -199,7 +198,7 @@ func testClient(t *testing.T, srv *Server) {
 	defer client.Close()
 	for _, test := range tests {
 		var res testData
-		reqURL := fmt.Sprintf(e.String() + test.path)
+		reqURL := e.String() + test.path
 		req, err := http.NewRequest(test.method, reqURL, nil)
 		if err != nil {
 			t.Fatal(err)
@@ -308,10 +307,6 @@ func TestTimeout(t *testing.T) {
 	}
 }
 
-func TestLogger(_ *testing.T) {
-	// TODO
-}
-
 func TestRequestDecoder(t *testing.T) {
 	o := &Server{}
 	v := func(*http.Request, interface{}) error { return nil }