Repository: zeromicro/go-zero
Pull request #2824 (2023-01-28 02:47:36)

feat: add MustNewRedis

codecov-commenter: # [Codecov](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#2824](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (17ee0ba) into [master](https://codecov.io/gh/zeromicro/go-zero/commit/ab9eeff5003cadbaadbefec8ead16d0f236512ce?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (ab9eeff) will **decrease** coverage by `0.25%`.
> The diff coverage is `20.00%`.

:mega: This organization is not using Codecov’s [GitHub App Integration](https://github.com/apps/codecov). We recommend you install it so Codecov can continue to function properly for your repositories. [Learn more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)


| [Impacted Files](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [core/stores/redis/conf.go](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-Y29yZS9zdG9yZXMvcmVkaXMvY29uZi5nbw==) | `100.00% <ø> (ø)` | |
| [core/stores/redis/redis.go](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-Y29yZS9zdG9yZXMvcmVkaXMvcmVkaXMuZ28=) | `97.52% <7.69%> (-1.51%)` | :arrow_down: |
| [core/stores/cache/cache.go](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-Y29yZS9zdG9yZXMvY2FjaGUvY2FjaGUuZ28=) | `92.94% <100.00%> (ø)` | |
| [core/stores/kv/store.go](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-Y29yZS9zdG9yZXMva3Yvc3RvcmUuZ28=) | `98.34% <100.00%> (ø)` | |
| [zrpc/server.go](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-enJwYy9zZXJ2ZXIuZ28=) | `78.87% <100.00%> (ø)` | |
| [core/stores/cache/cleaner.go](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-Y29yZS9zdG9yZXMvY2FjaGUvY2xlYW5lci5nbw==) | `46.51% <0.00%> (-32.56%)` | :arrow_down: |
| [core/stores/cache/cachenode.go](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-Y29yZS9zdG9yZXMvY2FjaGUvY2FjaGVub2RlLmdv) | `39.72% <0.00%> (-6.17%)` | :arrow_down: |
| [core/logx/rotatelogger.go](https://codecov.io/gh/zeromicro/go-zero/pull/2824?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-Y29yZS9sb2d4L3JvdGF0ZWxvZ2dlci5nbw==) | `73.80% <0.00%> (+1.10%)` | :arrow_up: |


Patch:
diff --git a/core/mr/mapreduce.go b/core/mr/mapreduce.go
index e5ab7041..edb32808 100644
--- a/core/mr/mapreduce.go
+++ b/core/mr/mapreduce.go
@@ -7,7 +7,6 @@ import (
 	"sync/atomic"
 
 	"github.com/zeromicro/go-zero/core/errorx"
-	"github.com/zeromicro/go-zero/core/lang"
 )
 
 const (
@@ -24,30 +23,30 @@ var (
 
 type (
 	// ForEachFunc is used to do element processing, but no output.
-	ForEachFunc func(item any)
+	ForEachFunc[T any] func(item T)
 	// GenerateFunc is used to let callers send elements into source.
-	GenerateFunc func(source chan<- any)
+	GenerateFunc[T any] func(source chan<- T)
 	// MapFunc is used to do element processing and write the output to writer.
-	MapFunc func(item any, writer Writer)
+	MapFunc[T, U any] func(item T, writer Writer[U])
 	// MapperFunc is used to do element processing and write the output to writer,
 	// use cancel func to cancel the processing.
-	MapperFunc func(item any, writer Writer, cancel func(error))
+	MapperFunc[T, U any] func(item T, writer Writer[U], cancel func(error))
 	// ReducerFunc is used to reduce all the mapping output and write to writer,
 	// use cancel func to cancel the processing.
-	ReducerFunc func(pipe <-chan any, writer Writer, cancel func(error))
+	ReducerFunc[U, V any] func(pipe <-chan U, writer Writer[V], cancel func(error))
 	// VoidReducerFunc is used to reduce all the mapping output, but no output.
 	// Use cancel func to cancel the processing.
-	VoidReducerFunc func(pipe <-chan any, cancel func(error))
+	VoidReducerFunc[U any] func(pipe <-chan U, cancel func(error))
 	// Option defines the method to customize the mapreduce.
 	Option func(opts *mapReduceOptions)
 
-	mapperContext struct {
+	mapperContext[T, U any] struct {
 		ctx       context.Context
-		mapper    MapFunc
-		source    <-chan any
+		mapper    MapFunc[T, U]
+		source    <-chan T
 		panicChan *onceChan
-		collector chan<- any
-		doneChan  <-chan lang.PlaceholderType
+		collector chan<- U
+		doneChan  <-chan struct{}
 		workers   int
 	}
 
@@ -57,8 +56,8 @@ type (
 	}
 
 	// Writer interface wraps Write method.
-	Writer interface {
-		Write(v any)
+	Writer[T any] interface {
+		Write(v T)
 	}
 )
 
@@ -68,12 +67,11 @@ func Finish(fns ...func() error) error {
 		return nil
 	}
 
-	return MapReduceVoid(func(source chan<- any) {
+	return MapReduceVoid(func(source chan<- func() error) {
 		for _, fn := range fns {
 			source <- fn
 		}
-	}, func(item any, writer Writer, cancel func(error)) {
-		fn := item.(func() error)
+	}, func(fn func() error, writer Writer[any], cancel func(error)) {
 		if err := fn(); err != nil {
 			cancel(err)
 		}
@@ -87,27 +85,26 @@ func FinishVoid(fns ...func()) {
 		return
 	}
 
-	ForEach(func(source chan<- any) {
+	ForEach(func(source chan<- func()) {
 		for _, fn := range fns {
 			source <- fn
 		}
-	}, func(item any) {
-		fn := item.(func())
+	}, func(fn func()) {
 		fn()
 	}, WithWorkers(len(fns)))
 }
 
 // ForEach maps all elements from given generate but no output.
-func ForEach(generate GenerateFunc, mapper ForEachFunc, opts ...Option) {
+func ForEach[T any](generate GenerateFunc[T], mapper ForEachFunc[T], opts ...Option) {
 	options := buildOptions(opts...)
 	panicChan := &onceChan{channel: make(chan any)}
 	source := buildSource(generate, panicChan)
 	collector := make(chan any)
-	done := make(chan lang.PlaceholderType)
+	done := make(chan struct{})
 
-	go executeMappers(mapperContext{
+	go executeMappers(mapperContext[T, any]{
 		ctx: options.ctx,
-		mapper: func(item any, _ Writer) {
+		mapper: func(item T, _ Writer[any]) {
 			mapper(item)
 		},
 		source:    source,
@@ -131,26 +128,26 @@ func ForEach(generate GenerateFunc, mapper ForEachFunc, opts ...Option) {
 
 // MapReduce maps all elements generated from given generate func,
 // and reduces the output elements with given reducer.
-func MapReduce(generate GenerateFunc, mapper MapperFunc, reducer ReducerFunc,
-	opts ...Option) (any, error) {
+func MapReduce[T, U, V any](generate GenerateFunc[T], mapper MapperFunc[T, U], reducer ReducerFunc[U, V],
+	opts ...Option) (V, error) {
 	panicChan := &onceChan{channel: make(chan any)}
 	source := buildSource(generate, panicChan)
 	return mapReduceWithPanicChan(source, panicChan, mapper, reducer, opts...)
 }
 
 // MapReduceChan maps all elements from source, and reduce the output elements with given reducer.
-func MapReduceChan(source <-chan any, mapper MapperFunc, reducer ReducerFunc,
-	opts ...Option) (any, error) {
+func MapReduceChan[T, U, V any](source <-chan T, mapper MapperFunc[T, U], reducer ReducerFunc[U, V],
+	opts ...Option) (V, error) {
 	panicChan := &onceChan{channel: make(chan any)}
 	return mapReduceWithPanicChan(source, panicChan, mapper, reducer, opts...)
 }
 
 // mapReduceWithPanicChan maps all elements from source, and reduce the output elements with given reducer.
-func mapReduceWithPanicChan(source <-chan any, panicChan *onceChan, mapper MapperFunc,
-	reducer ReducerFunc, opts ...Option) (any, error) {
+func mapReduceWithPanicChan[T, U, V any](source <-chan T, panicChan *onceChan, mapper MapperFunc[T, U],
+	reducer ReducerFunc[U, V], opts ...Option) (val V, err error) {
 	options := buildOptions(opts...)
 	// output is used to write the final result
-	output := make(chan any)
+	output := make(chan V)
 	defer func() {
 		// reducer can only write once, if more, panic
 		for range output {
@@ -159,12 +156,12 @@ func mapReduceWithPanicChan(source <-chan any, panicChan *onceChan, mapper Mappe
 	}()
 
 	// collector is used to collect data from mapper, and consume in reducer
-	collector := make(chan any, options.workers)
+	collector := make(chan U, options.workers)
 	// if done is closed, all mappers and reducer should stop processing
-	done := make(chan lang.PlaceholderType)
+	done := make(chan struct{})
 	writer := newGuardedWriter(options.ctx, output, done)
 	var closeOnce sync.Once
-	// use atomic.Value to avoid data race
+	// use atomic type to avoid data race
 	var retErr errorx.AtomicError
 	finish := func() {
 		closeOnce.Do(func() {
@@ -195,9 +192,9 @@ func mapReduceWithPanicChan(source <-chan any, panicChan *onceChan, mapper Mappe
 		reducer(collector, writer, cancel)
 	}()
 
-	go executeMappers(mapperContext{
+	go executeMappers(mapperContext[T, U]{
 		ctx: options.ctx,
-		mapper: func(item any, w Writer) {
+		mapper: func(item T, w Writer[U]) {
 			mapper(item, w, cancel)
 		},
 		source:    source,
@@ -210,26 +207,29 @@ func mapReduceWithPanicChan(source <-chan any, panicChan *onceChan, mapper Mappe
 	select {
 	case <-options.ctx.Done():
 		cancel(context.DeadlineExceeded)
-		return nil, context.DeadlineExceeded
+		err = context.DeadlineExceeded
 	case v := <-panicChan.channel:
 		// drain output here, otherwise for loop panic in defer
 		drain(output)
 		panic(v)
 	case v, ok := <-output:
-		if err := retErr.Load(); err != nil {
-			return nil, err
+		if e := retErr.Load(); e != nil {
+			err = e
 		} else if ok {
-			return v, nil
+			val = v
 		} else {
-			return nil, ErrReduceNoOutput
+			err = ErrReduceNoOutput
 		}
 	}
+
+	return
 }
 
 // MapReduceVoid maps all elements generated from given generate,
 // and reduce the output elements with given reducer.
-func MapReduceVoid(generate GenerateFunc, mapper MapperFunc, reducer VoidReducerFunc, opts ...Option) error {
-	_, err := MapReduce(generate, mapper, func(input <-chan any, writer Writer, cancel func(error)) {
+func MapReduceVoid[T, U any](generate GenerateFunc[T], mapper MapperFunc[T, U],
+	reducer VoidReducerFunc[U], opts ...Option) error {
+	_, err := MapReduce(generate, mapper, func(input <-chan U, writer Writer[any], cancel func(error)) {
 		reducer(input, cancel)
 	}, opts...)
 	if errors.Is(err, ErrReduceNoOutput) {
@@ -266,8 +266,8 @@ func buildOptions(opts ...Option) *mapReduceOptions {
 	return options
 }
 
-func buildSource(generate GenerateFunc, panicChan *onceChan) chan any {
-	source := make(chan any)
+func buildSource[T any](generate GenerateFunc[T], panicChan *onceChan) chan T {
+	source := make(chan T)
 	go func() {
 		defer func() {
 			if r := recover(); r != nil {
@@ -283,13 +283,13 @@ func buildSource(generate GenerateFunc, panicChan *onceChan) chan any {
 }
 
 // drain drains the channel.
-func drain(channel <-chan any) {
+func drain[T any](channel <-chan T) {
 	// drain the channel
 	for range channel {
 	}
 }
 
-func executeMappers(mCtx mapperContext) {
+func executeMappers[T, U any](mCtx mapperContext[T, U]) {
 	var wg sync.WaitGroup
 	defer func() {
 		wg.Wait()
@@ -298,7 +298,7 @@ func executeMappers(mCtx mapperContext) {
 	}()
 
 	var failed int32
-	pool := make(chan lang.PlaceholderType, mCtx.workers)
+	pool := make(chan struct{}, mCtx.workers)
 	writer := newGuardedWriter(mCtx.ctx, mCtx.collector, mCtx.doneChan)
 	for atomic.LoadInt32(&failed) == 0 {
 		select {
@@ -306,7 +306,7 @@ func executeMappers(mCtx mapperContext) {
 			return
 		case <-mCtx.doneChan:
 			return
-		case pool <- lang.Placeholder:
+		case pool <- struct{}{}:
 			item, ok := <-mCtx.source
 			if !ok {
 				<-pool
@@ -346,22 +346,21 @@ func once(fn func(error)) func(error) {
 	}
 }
 
-type guardedWriter struct {
+type guardedWriter[T any] struct {
 	ctx     context.Context
-	channel chan<- any
-	done    <-chan lang.PlaceholderType
+	channel chan<- T
+	done    <-chan struct{}
 }
 
-func newGuardedWriter(ctx context.Context, channel chan<- any,
-	done <-chan lang.PlaceholderType) guardedWriter {
-	return guardedWriter{
+func newGuardedWriter[T any](ctx context.Context, channel chan<- T, done <-chan struct{}) guardedWriter[T] {
+	return guardedWriter[T]{
 		ctx:     ctx,
 		channel: channel,
 		done:    done,
 	}
 }
 
-func (gw guardedWriter) Write(v any) {
+func (gw guardedWriter[T]) Write(v T) {
 	select {
 	case <-gw.ctx.Done():
 		return
diff --git a/core/mr/readme-cn.md b/core/mr/readme-cn.md
index 888868ca..7aaf3006 100644
--- a/core/mr/readme-cn.md
+++ b/core/mr/readme-cn.md
@@ -58,20 +58,19 @@ import (
 )
 
 func main() {
-    val, err := mr.MapReduce(func(source chan<- any) {
+    val, err := mr.MapReduce(func(source chan<- int) {
         // generator
         for i := 0; i < 10; i++ {
             source <- i
         }
-    }, func(item any, writer mr.Writer, cancel func(error)) {
+    }, func(i int, writer mr.Writer[int], cancel func(error)) {
         // mapper
-        i := item.(int)
         writer.Write(i * i)
-    }, func(pipe <-chan any, writer mr.Writer, cancel func(error)) {
+    }, func(pipe <-chan int, writer mr.Writer[int], cancel func(error)) {
         // reducer
         var sum int
         for i := range pipe {
-            sum += i.(int)
+            sum += i
         }
         writer.Write(sum)
     })
diff --git a/core/mr/readme.md b/core/mr/readme.md
index ba719a6c..3bb916b5 100644
--- a/core/mr/readme.md
+++ b/core/mr/readme.md
@@ -59,20 +59,19 @@ import (
 )
 
 func main() {
-    val, err := mr.MapReduce(func(source chan<- any) {
+    val, err := mr.MapReduce(func(source chan<- int) {
         // generator
         for i := 0; i < 10; i++ {
             source <- i
         }
-    }, func(item any, writer mr.Writer, cancel func(error)) {
+    }, func(i int, writer mr.Writer[int], cancel func(error)) {
         // mapper
-        i := item.(int)
         writer.Write(i * i)
-    }, func(pipe <-chan any, writer mr.Writer, cancel func(error)) {
+    }, func(pipe <-chan int, writer mr.Writer[int], cancel func(error)) {
         // reducer
         var sum int
         for i := range pipe {
-            sum += i.(int)
+            sum += i
         }
         writer.Write(sum)
     })
diff --git a/core/stores/cache/cache.go b/core/stores/cache/cache.go
index d9b30b7f..edec63bc 100644
--- a/core/stores/cache/cache.go
+++ b/core/stores/cache/cache.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/zeromicro/go-zero/core/errorx"
 	"github.com/zeromicro/go-zero/core/hash"
+	"github.com/zeromicro/go-zero/core/stores/redis"
 	"github.com/zeromicro/go-zero/core/syncx"
 )
 
@@ -62,12 +63,12 @@ func New(c ClusterConf, barrier syncx.SingleFlight, st *Stat, errNotFound error,
 	}
 
 	if len(c) == 1 {
-		return NewNode(c[0].NewRedis(), barrier, st, errNotFound, opts...)
+		return NewNode(redis.MustNewRedis(c[0].RedisConf), barrier, st, errNotFound, opts...)
 	}
 
 	dispatcher := hash.NewConsistentHash()
 	for _, node := range c {
-		cn := NewNode(node.NewRedis(), barrier, st, errNotFound, opts...)
+		cn := NewNode(redis.MustNewRedis(node.RedisConf), barrier, st, errNotFound, opts...)
 		dispatcher.AddWithWeight(cn, node.Weight)
 	}
 
diff --git a/core/stores/kv/store.go b/core/stores/kv/store.go
index ca9d113f..7b500302 100644
--- a/core/stores/kv/store.go
+++ b/core/stores/kv/store.go
@@ -164,7 +164,7 @@ func NewStore(c KvConf) Store {
 	// because Store and redis.Redis has different methods.
 	dispatcher := hash.NewConsistentHash()
 	for _, node := range c {
-		cn := node.NewRedis()
+		cn := redis.MustNewRedis(node.RedisConf)
 		dispatcher.AddWithWeight(cn, node.Weight)
 	}
 
diff --git a/core/stores/redis/conf.go b/core/stores/redis/conf.go
index f8ee90c6..711ce817 100644
--- a/core/stores/redis/conf.go
+++ b/core/stores/redis/conf.go
@@ -9,6 +9,8 @@ var (
 	ErrEmptyType = errors.New("empty redis type")
 	// ErrEmptyKey is an error that indicates no redis key is set.
 	ErrEmptyKey = errors.New("empty redis key")
+	// ErrPing is an error that indicates ping failed.
+	ErrPing = errors.New("ping redis failed")
 )
 
 type (
@@ -27,7 +29,7 @@ type (
 	}
 )
 
-// NewRedis returns a Redis.
+// Deprecated: use MustNewRedis or NewRedis instead.
 func (rc RedisConf) NewRedis() *Redis {
 	var opts []Option
 	if rc.Type == ClusterType {
diff --git a/core/stores/redis/redis.go b/core/stores/redis/redis.go
index dcec2a7a..9572aeed 100644
--- a/core/stores/redis/redis.go
+++ b/core/stores/redis/redis.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"strconv"
 	"time"
 
@@ -85,8 +86,46 @@ type (
 	StringCmd = red.StringCmd
 )
 
-// New returns a Redis with given options.
+// Deprecated: use MustNewRedis or NewRedis instead.
 func New(addr string, opts ...Option) *Redis {
+	return newRedis(addr, opts...)
+}
+
+// MustNewRedis returns a Redis with given options.
+func MustNewRedis(conf RedisConf, opts ...Option) *Redis {
+	rds, err := NewRedis(conf, opts...)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return rds
+}
+
+// NewRedis returns a Redis with given options.
+func NewRedis(conf RedisConf, opts ...Option) (*Redis, error) {
+	if err := conf.Validate(); err != nil {
+		return nil, err
+	}
+
+	if conf.Type == ClusterType {
+		opts = append([]Option{Cluster()}, opts...)
+	}
+	if len(conf.Pass) > 0 {
+		opts = append([]Option{WithPass(conf.Pass)}, opts...)
+	}
+	if conf.Tls {
+		opts = append([]Option{WithTLS()}, opts...)
+	}
+
+	rds := newRedis(conf.Host, opts...)
+	if !rds.Ping() {
+		return nil, ErrPing
+	}
+
+	return rds, nil
+}
+
+func newRedis(addr string, opts ...Option) *Redis {
 	r := &Redis{
 		Addr: addr,
 		Type: NodeType,
diff --git a/gateway/server.go b/gateway/server.go
index 489796e0..c6972709 100644
--- a/gateway/server.go
+++ b/gateway/server.go
@@ -63,12 +63,11 @@ func (s *Server) build() error {
 		return err
 	}
 
-	return mr.MapReduceVoid(func(source chan<- any) {
+	return mr.MapReduceVoid(func(source chan<- Upstream) {
 		for _, up := range s.upstreams {
 			source <- up
 		}
-	}, func(item any, writer mr.Writer, cancel func(error)) {
-		up := item.(Upstream)
+	}, func(up Upstream, writer mr.Writer[rest.Route], cancel func(error)) {
 		cli := zrpc.MustNewClient(up.Grpc)
 		source, err := s.createDescriptorSource(cli, up)
 		if err != nil {
@@ -109,9 +108,8 @@ func (s *Server) build() error {
 				Handler: s.buildHandler(source, resolver, cli, m.RpcPath),
 			})
 		}
-	}, func(pipe <-chan any, cancel func(error)) {
-		for item := range pipe {
-			route := item.(rest.Route)
+	}, func(pipe <-chan rest.Route, cancel func(error)) {
+		for route := range pipe {
 			s.Server.AddRoute(route)
 		}
 	})
diff --git a/internal/health/health.go b/internal/health/health.go
index fd4c8811..7c238d9c 100644
--- a/internal/health/health.go
+++ b/internal/health/health.go
@@ -9,7 +9,7 @@ import (
 	"github.com/zeromicro/go-zero/core/syncx"
 )
 
-// defaultHealthManager is global comboHealthManager for byone self.
+// defaultHealthManager is global comboHealthManager.
 var defaultHealthManager = newComboHealthManager()
 
 type (
diff --git a/zrpc/server.go b/zrpc/server.go
index 99a637ba..5327b83e 100644
--- a/zrpc/server.go
+++ b/zrpc/server.go
@@ -7,6 +7,7 @@ import (
 	"github.com/zeromicro/go-zero/core/load"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stat"
+	"github.com/zeromicro/go-zero/core/stores/redis"
 	"github.com/zeromicro/go-zero/zrpc/internal"
 	"github.com/zeromicro/go-zero/zrpc/internal/auth"
 	"github.com/zeromicro/go-zero/zrpc/internal/serverinterceptors"
@@ -120,7 +121,12 @@ func setupInterceptors(server internal.Server, c RpcServerConf, metrics *stat.Me
 	}
 
 	if c.Auth {
-		authenticator, err := auth.NewAuthenticator(c.Redis.NewRedis(), c.Redis.Key, c.StrictControl)
+		rds, err := redis.NewRedis(c.Redis.RedisConf)
+		if err != nil {
+			return err
+		}
+
+		authenticator, err := auth.NewAuthenticator(rds, c.Redis.Key, c.StrictControl)
 		if err != nil {
 			return err
 		}


Test patch:
diff --git a/core/mr/mapreduce_fuzz_test.go b/core/mr/mapreduce_fuzz_test.go
index 15199808..936d3623 100644
--- a/core/mr/mapreduce_fuzz_test.go
+++ b/core/mr/mapreduce_fuzz_test.go
@@ -1,6 +1,3 @@
-//go:build go1.18
-// +build go1.18
-
 package mr
 
 import (
@@ -18,9 +15,9 @@ import (
 func FuzzMapReduce(f *testing.F) {
 	rand.Seed(time.Now().UnixNano())
 
-	f.Add(uint(10), uint(runtime.NumCPU()))
-	f.Fuzz(func(t *testing.T, num, workers uint) {
-		n := int64(num)%5000 + 5000
+	f.Add(int64(10), runtime.NumCPU())
+	f.Fuzz(func(t *testing.T, n int64, workers int) {
+		n = n%5000 + 5000
 		genPanic := rand.Intn(100) == 0
 		mapperPanic := rand.Intn(100) == 0
 		reducerPanic := rand.Intn(100) == 0
@@ -29,34 +26,33 @@ func FuzzMapReduce(f *testing.F) {
 		reducerIdx := rand.Int63n(n)
 		squareSum := (n - 1) * n * (2*n - 1) / 6
 
-		fn := func() (any, error) {
+		fn := func() (int64, error) {
 			defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
 
-			return MapReduce(func(source chan<- any) {
+			return MapReduce(func(source chan<- int64) {
 				for i := int64(0); i < n; i++ {
 					source <- i
 					if genPanic && i == genIdx {
 						panic("foo")
 					}
 				}
-			}, func(item any, writer Writer, cancel func(error)) {
-				v := item.(int64)
+			}, func(v int64, writer Writer[int64], cancel func(error)) {
 				if mapperPanic && v == mapperIdx {
 					panic("bar")
 				}
 				writer.Write(v * v)
-			}, func(pipe <-chan any, writer Writer, cancel func(error)) {
+			}, func(pipe <-chan int64, writer Writer[int64], cancel func(error)) {
 				var idx int64
 				var total int64
 				for v := range pipe {
 					if reducerPanic && idx == reducerIdx {
 						panic("baz")
 					}
-					total += v.(int64)
+					total += v
 					idx++
 				}
 				writer.Write(total)
-			}, WithWorkers(int(workers)%50+runtime.NumCPU()/2))
+			}, WithWorkers(workers%50+runtime.NumCPU()))
 		}
 
 		if genPanic || mapperPanic || reducerPanic {
@@ -72,7 +68,7 @@ func FuzzMapReduce(f *testing.F) {
 		} else {
 			val, err := fn()
 			assert.Nil(t, err)
-			assert.Equal(t, squareSum, val.(int64))
+			assert.Equal(t, squareSum, val)
 		}
 	})
 }
diff --git a/core/mr/mapreduce_fuzzcase_test.go b/core/mr/mapreduce_fuzzcase_test.go
index 39b3ab39..7ce2b58f 100644
--- a/core/mr/mapreduce_fuzzcase_test.go
+++ b/core/mr/mapreduce_fuzzcase_test.go
@@ -54,28 +54,27 @@ func TestMapReduceRandom(t *testing.T) {
 				reducerIdx := rand.Int63n(n)
 				squareSum := (n - 1) * n * (2*n - 1) / 6
 
-				fn := func() (any, error) {
-					return MapReduce(func(source chan<- any) {
+				fn := func() (int64, error) {
+					return MapReduce(func(source chan<- int64) {
 						for i := int64(0); i < n; i++ {
 							source <- i
 							if genPanic && i == genIdx {
 								panic("foo")
 							}
 						}
-					}, func(item any, writer Writer, cancel func(error)) {
-						v := item.(int64)
+					}, func(v int64, writer Writer[int64], cancel func(error)) {
 						if mapperPanic && v == mapperIdx {
 							panic("bar")
 						}
 						writer.Write(v * v)
-					}, func(pipe <-chan any, writer Writer, cancel func(error)) {
+					}, func(pipe <-chan int64, writer Writer[int64], cancel func(error)) {
 						var idx int64
 						var total int64
 						for v := range pipe {
 							if reducerPanic && idx == reducerIdx {
 								panic("baz")
 							}
-							total += v.(int64)
+							total += v
 							idx++
 						}
 						writer.Write(total)
@@ -95,7 +94,7 @@ func TestMapReduceRandom(t *testing.T) {
 				} else {
 					val, err := fn()
 					assert.Nil(t, err)
-					assert.Equal(t, squareSum, val.(int64))
+					assert.Equal(t, squareSum, val)
 				}
 				bar.Increment()
 			})
diff --git a/core/mr/mapreduce_test.go b/core/mr/mapreduce_test.go
index 79030fc7..00418cc8 100644
--- a/core/mr/mapreduce_test.go
+++ b/core/mr/mapreduce_test.go
@@ -3,7 +3,7 @@ package mr
 import (
 	"context"
 	"errors"
-	"io"
+	"io/ioutil"
 	"log"
 	"runtime"
 	"sync/atomic"
@@ -17,7 +17,7 @@ import (
 var errDummy = errors.New("dummy")
 
 func init() {
-	log.SetOutput(io.Discard)
+	log.SetOutput(ioutil.Discard)
 }
 
 func TestFinish(t *testing.T) {
@@ -91,11 +91,11 @@ func TestForEach(t *testing.T) {
 		defer goleak.VerifyNone(t)
 
 		var count uint32
-		ForEach(func(source chan<- any) {
+		ForEach(func(source chan<- int) {
 			for i := 0; i < tasks; i++ {
 				source <- i
 			}
-		}, func(item any) {
+		}, func(item int) {
 			atomic.AddUint32(&count, 1)
 		}, WithWorkers(-1))
 
@@ -106,12 +106,12 @@ func TestForEach(t *testing.T) {
 		defer goleak.VerifyNone(t)
 
 		var count uint32
-		ForEach(func(source chan<- any) {
+		ForEach(func(source chan<- int) {
 			for i := 0; i < tasks; i++ {
 				source <- i
 			}
-		}, func(item any) {
-			if item.(int)%2 == 0 {
+		}, func(item int) {
+			if item%2 == 0 {
 				atomic.AddUint32(&count, 1)
 			}
 		})
@@ -123,11 +123,11 @@ func TestForEach(t *testing.T) {
 		defer goleak.VerifyNone(t)
 
 		assert.PanicsWithValue(t, "foo", func() {
-			ForEach(func(source chan<- any) {
+			ForEach(func(source chan<- int) {
 				for i := 0; i < tasks; i++ {
 					source <- i
 				}
-			}, func(item any) {
+			}, func(item int) {
 				panic("foo")
 			})
 		})
@@ -139,9 +139,9 @@ func TestGeneratePanic(t *testing.T) {
 
 	t.Run("all", func(t *testing.T) {
 		assert.PanicsWithValue(t, "foo", func() {
-			ForEach(func(source chan<- any) {
+			ForEach(func(source chan<- int) {
 				panic("foo")
-			}, func(item any) {
+			}, func(item int) {
 			})
 		})
 	})
@@ -154,14 +154,14 @@ func TestMapperPanic(t *testing.T) {
 	var run int32
 	t.Run("all", func(t *testing.T) {
 		assert.PanicsWithValue(t, "foo", func() {
-			_, _ = MapReduce(func(source chan<- any) {
+			_, _ = MapReduce(func(source chan<- int) {
 				for i := 0; i < tasks; i++ {
 					source <- i
 				}
-			}, func(item any, writer Writer, cancel func(error)) {
+			}, func(item int, writer Writer[int], cancel func(error)) {
 				atomic.AddInt32(&run, 1)
 				panic("foo")
-			}, func(pipe <-chan any, writer Writer, cancel func(error)) {
+			}, func(pipe <-chan int, writer Writer[int], cancel func(error)) {
 			})
 		})
 		assert.True(t, atomic.LoadInt32(&run) < tasks/2)
@@ -173,10 +173,10 @@ func TestMapReduce(t *testing.T) {
 
 	tests := []struct {
 		name        string
-		mapper      MapperFunc
-		reducer     ReducerFunc
+		mapper      MapperFunc[int, int]
+		reducer     ReducerFunc[int, int]
 		expectErr   error
-		expectValue any
+		expectValue int
 	}{
 		{
 			name:        "simple",
@@ -185,8 +185,7 @@ func TestMapReduce(t *testing.T) {
 		},
 		{
 			name: "cancel with error",
-			mapper: func(item any, writer Writer, cancel func(error)) {
-				v := item.(int)
+			mapper: func(v int, writer Writer[int], cancel func(error)) {
 				if v%3 == 0 {
 					cancel(errDummy)
 				}
@@ -196,22 +195,20 @@ func TestMapReduce(t *testing.T) {
 		},
 		{
 			name: "cancel with nil",
-			mapper: func(item any, writer Writer, cancel func(error)) {
-				v := item.(int)
+			mapper: func(v int, writer Writer[int], cancel func(error)) {
 				if v%3 == 0 {
 					cancel(nil)
 				}
 				writer.Write(v * v)
 			},
-			expectErr:   ErrCancelWithNil,
-			expectValue: nil,
+			expectErr: ErrCancelWithNil,
 		},
 		{
 			name: "cancel with more",
-			reducer: func(pipe <-chan any, writer Writer, cancel func(error)) {
+			reducer: func(pipe <-chan int, writer Writer[int], cancel func(error)) {
 				var result int
 				for item := range pipe {
-					result += item.(int)
+					result += item
 					if result > 10 {
 						cancel(errDummy)
 					}
@@ -226,21 +223,20 @@ func TestMapReduce(t *testing.T) {
 		for _, test := range tests {
 			t.Run(test.name, func(t *testing.T) {
 				if test.mapper == nil {
-					test.mapper = func(item any, writer Writer, cancel func(error)) {
-						v := item.(int)
+					test.mapper = func(v int, writer Writer[int], cancel func(error)) {
 						writer.Write(v * v)
 					}
 				}
 				if test.reducer == nil {
-					test.reducer = func(pipe <-chan any, writer Writer, cancel func(error)) {
+					test.reducer = func(pipe <-chan int, writer Writer[int], cancel func(error)) {
 						var result int
 						for item := range pipe {
-							result += item.(int)
+							result += item
 						}
 						writer.Write(result)
 					}
 				}
-				value, err := MapReduce(func(source chan<- any) {
+				value, err := MapReduce(func(source chan<- int) {
 					for i := 1; i < 5; i++ {
 						source <- i
 					}
@@ -256,22 +252,21 @@ func TestMapReduce(t *testing.T) {
 		for _, test := range tests {
 			t.Run(test.name, func(t *testing.T) {
 				if test.mapper == nil {
-					test.mapper = func(item any, writer Writer, cancel func(error)) {
-						v := item.(int)
+					test.mapper = func(v int, writer Writer[int], cancel func(error)) {
 						writer.Write(v * v)
 					}
 				}
 				if test.reducer == nil {
-					test.reducer = func(pipe <-chan any, writer Writer, cancel func(error)) {
+					test.reducer = func(pipe <-chan int, writer Writer[int], cancel func(error)) {
 						var result int
 						for item := range pipe {
-							result += item.(int)
+							result += item
 						}
 						writer.Write(result)
 					}
 				}
 
-				source := make(chan any)
+				source := make(chan int)
 				go func() {
 					for i := 1; i < 5; i++ {
 						source <- i
@@ -291,13 +286,13 @@ func TestMapReduceWithReduerWriteMoreThanOnce(t *testing.T) {
 	defer goleak.VerifyNone(t)
 
 	assert.Panics(t, func() {
-		MapReduce(func(source chan<- any) {
+		MapReduce(func(source chan<- int) {
 			for i := 0; i < 10; i++ {
 				source <- i
 			}
-		}, func(item any, writer Writer, cancel func(error)) {
+		}, func(item int, writer Writer[int], cancel func(error)) {
 			writer.Write(item)
-		}, func(pipe <-chan any, writer Writer, cancel func(error)) {
+		}, func(pipe <-chan int, writer Writer[string], cancel func(error)) {
 			drain(pipe)
 			writer.Write("one")
 			writer.Write("two")
@@ -311,8 +306,8 @@ func TestMapReduceVoid(t *testing.T) {
 	var value uint32
 	tests := []struct {
 		name        string
-		mapper      MapperFunc
-		reducer     VoidReducerFunc
+		mapper      MapperFunc[int, int]
+		reducer     VoidReducerFunc[int]
 		expectValue uint32
 		expectErr   error
 	}{
@@ -323,8 +318,7 @@ func TestMapReduceVoid(t *testing.T) {
 		},
 		{
 			name: "cancel with error",
-			mapper: func(item any, writer Writer, cancel func(error)) {
-				v := item.(int)
+			mapper: func(v int, writer Writer[int], cancel func(error)) {
 				if v%3 == 0 {
 					cancel(errDummy)
 				}
@@ -334,8 +328,7 @@ func TestMapReduceVoid(t *testing.T) {
 		},
 		{
 			name: "cancel with nil",
-			mapper: func(item any, writer Writer, cancel func(error)) {
-				v := item.(int)
+			mapper: func(v int, writer Writer[int], cancel func(error)) {
 				if v%3 == 0 {
 					cancel(nil)
 				}
@@ -345,9 +338,9 @@ func TestMapReduceVoid(t *testing.T) {
 		},
 		{
 			name: "cancel with more",
-			reducer: func(pipe <-chan any, cancel func(error)) {
+			reducer: func(pipe <-chan int, cancel func(error)) {
 				for item := range pipe {
-					result := atomic.AddUint32(&value, uint32(item.(int)))
+					result := atomic.AddUint32(&value, uint32(item))
 					if result > 10 {
 						cancel(errDummy)
 					}
@@ -362,19 +355,18 @@ func TestMapReduceVoid(t *testing.T) {
 			atomic.StoreUint32(&value, 0)
 
 			if test.mapper == nil {
-				test.mapper = func(item any, writer Writer, cancel func(error)) {
-					v := item.(int)
+				test.mapper = func(v int, writer Writer[int], cancel func(error)) {
 					writer.Write(v * v)
 				}
 			}
 			if test.reducer == nil {
-				test.reducer = func(pipe <-chan any, cancel func(error)) {
+				test.reducer = func(pipe <-chan int, cancel func(error)) {
 					for item := range pipe {
-						atomic.AddUint32(&value, uint32(item.(int)))
+						atomic.AddUint32(&value, uint32(item))
 					}
 				}
 			}
-			err := MapReduceVoid(func(source chan<- any) {
+			err := MapReduceVoid(func(source chan<- int) {
 				for i := 1; i < 5; i++ {
 					source <- i
 				}
@@ -392,18 +384,17 @@ func TestMapReduceVoidWithDelay(t *testing.T) {
 	defer goleak.VerifyNone(t)
 
 	var result []int
-	err := MapReduceVoid(func(source chan<- any) {
+	err := MapReduceVoid(func(source chan<- int) {
 		source <- 0
 		source <- 1
-	}, func(item any, writer Writer, cancel func(error)) {
-		i := item.(int)
+	}, func(i int, writer Writer[int], cancel func(error)) {
 		if i == 0 {
 			time.Sleep(time.Millisecond * 50)
 		}
 		writer.Write(i)
-	}, func(pipe <-chan any, cancel func(error)) {
+	}, func(pipe <-chan int, cancel func(error)) {
 		for item := range pipe {
-			i := item.(int)
+			i := item
 			result = append(result, i)
 		}
 	})
@@ -417,13 +408,12 @@ func TestMapReducePanic(t *testing.T) {
 	defer goleak.VerifyNone(t)
 
 	assert.Panics(t, func() {
-		_, _ = MapReduce(func(source chan<- any) {
+		_, _ = MapReduce(func(source chan<- int) {
 			source <- 0
 			source <- 1
-		}, func(item any, writer Writer, cancel func(error)) {
-			i := item.(int)
+		}, func(i int, writer Writer[int], cancel func(error)) {
 			writer.Write(i)
-		}, func(pipe <-chan any, writer Writer, cancel func(error)) {
+		}, func(pipe <-chan int, writer Writer[int], cancel func(error)) {
 			for range pipe {
 				panic("panic")
 			}
@@ -435,17 +425,16 @@ func TestMapReducePanicOnce(t *testing.T) {
 	defer goleak.VerifyNone(t)
 
 	assert.Panics(t, func() {
-		_, _ = MapReduce(func(source chan<- any) {
+		_, _ = MapReduce(func(source chan<- int) {
 			for i := 0; i < 100; i++ {
 				source <- i
 			}
-		}, func(item any, writer Writer, cancel func(error)) {
-			i := item.(int)
+		}, func(i int, writer Writer[int], cancel func(error)) {
 			if i == 0 {
 				panic("foo")
 			}
 			writer.Write(i)
-		}, func(pipe <-chan any, writer Writer, cancel func(error)) {
+		}, func(pipe <-chan int, writer Writer[int], cancel func(error)) {
 			for range pipe {
 				panic("bar")
 			}
@@ -457,12 +446,12 @@ func TestMapReducePanicBothMapperAndReducer(t *testing.T) {
 	defer goleak.VerifyNone(t)
 
 	assert.Panics(t, func() {
-		_, _ = MapReduce(func(source chan<- any) {
+		_, _ = MapReduce(func(source chan<- int) {
 			source <- 0
 			source <- 1
-		}, func(item any, writer Writer, cancel func(error)) {
+		}, func(item int, writer Writer[int], cancel func(error)) {
 			panic("foo")
-		}, func(pipe <-chan any, writer Writer, cancel func(error)) {
+		}, func(pipe <-chan int, writer Writer[int], cancel func(error)) {
 			panic("bar")
 		})
 	})
@@ -472,18 +461,17 @@ func TestMapReduceVoidCancel(t *testing.T) {
 	defer goleak.VerifyNone(t)
 
 	var result []int
-	err := MapReduceVoid(func(source chan<- any) {
+	err := MapReduceVoid(func(source chan<- int) {
 		source <- 0
 		source <- 1
-	}, func(item any, writer Writer, cancel func(error)) {
-		i := item.(int)
+	}, func(i int, writer Writer[int], cancel func(error)) {
 		if i == 1 {
 			cancel(errors.New("anything"))
 		}
 		writer.Write(i)
-	}, func(pipe <-chan any, cancel func(error)) {
+	}, func(pipe <-chan int, cancel func(error)) {
 		for item := range pipe {
-			i := item.(int)
+			i := item
 			result = append(result, i)
 		}
 	})
@@ -496,21 +484,19 @@ func TestMapReduceVoidCancelWithRemains(t *testing.T) {
 
 	var done int32
 	var result []int
-	err := MapReduceVoid(func(source chan<- any) {
+	err := MapReduceVoid(func(source chan<- int) {
 		for i := 0; i < defaultWorkers*2; i++ {
 			source <- i
 		}
 		atomic.AddInt32(&done, 1)
-	}, func(item any, writer Writer, cancel func(error)) {
-		i := item.(int)
+	}, func(i int, writer Writer[int], cancel func(error)) {
 		if i == defaultWorkers/2 {
 			cancel(errors.New("anything"))
 		}
 		writer.Write(i)
-	}, func(pipe <-chan any, cancel func(error)) {
+	}, func(pipe <-chan int, cancel func(error)) {
 		for item := range pipe {
-			i := item.(int)
-			result = append(result, i)
+			result = append(result, item)
 		}
 	})
 	assert.NotNil(t, err)
@@ -522,18 +508,18 @@ func TestMapReduceWithoutReducerWrite(t *testing.T) {
 	defer goleak.VerifyNone(t)
 
 	uids := []int{1, 2, 3}
-	res, err := MapReduce(func(source chan<- any) {
+	res, err := MapReduce(func(source chan<- int) {
 		for _, uid := range uids {
 			source <- uid
 		}
-	}, func(item any, writer Writer, cancel func(error)) {
+	}, func(item int, writer Writer[int], cancel func(error)) {
 		writer.Write(item)
-	}, func(pipe <-chan any, writer Writer, cancel func(error)) {
+	}, func(pipe <-chan int, writer Writer[int], cancel func(error)) {
 		drain(pipe)
 		// not calling writer.Write(...), should not panic
 	})
 	assert.Equal(t, ErrReduceNoOutput, err)
-	assert.Nil(t, res)
+	assert.Equal(t, 0, res)
 }
 
 func TestMapReduceVoidPanicInReducer(t *testing.T) {
@@ -542,15 +528,14 @@ func TestMapReduceVoidPanicInReducer(t *testing.T) {
 	const message = "foo"
 	assert.Panics(t, func() {
 		var done int32
-		_ = MapReduceVoid(func(source chan<- any) {
+		_ = MapReduceVoid(func(source chan<- int) {
 			for i := 0; i < defaultWorkers*2; i++ {
 				source <- i
 			}
 			atomic.AddInt32(&done, 1)
-		}, func(item any, writer Writer, cancel func(error)) {
-			i := item.(int)
+		}, func(i int, writer Writer[int], cancel func(error)) {
 			writer.Write(i)
-		}, func(pipe <-chan any, cancel func(error)) {
+		}, func(pipe <-chan int, cancel func(error)) {
 			panic(message)
 		}, WithWorkers(1))
 	})
@@ -561,13 +546,12 @@ func TestForEachWithContext(t *testing.T) {
 
 	var done int32
 	ctx, cancel := context.WithCancel(context.Background())
-	ForEach(func(source chan<- any) {
+	ForEach(func(source chan<- int) {
 		for i := 0; i < defaultWorkers*2; i++ {
 			source <- i
 		}
 		atomic.AddInt32(&done, 1)
-	}, func(item any) {
-		i := item.(int)
+	}, func(i int) {
 		if i == defaultWorkers/2 {
 			cancel()
 		}
@@ -580,20 +564,19 @@ func TestMapReduceWithContext(t *testing.T) {
 	var done int32
 	var result []int
 	ctx, cancel := context.WithCancel(context.Background())
-	err := MapReduceVoid(func(source chan<- any) {
+	err := MapReduceVoid(func(source chan<- int) {
 		for i := 0; i < defaultWorkers*2; i++ {
 			source <- i
 		}
 		atomic.AddInt32(&done, 1)
-	}, func(item any, writer Writer, c func(error)) {
-		i := item.(int)
+	}, func(i int, writer Writer[int], c func(error)) {
 		if i == defaultWorkers/2 {
 			cancel()
 		}
 		writer.Write(i)
-	}, func(pipe <-chan any, cancel func(error)) {
+	}, func(pipe <-chan int, cancel func(error)) {
 		for item := range pipe {
-			i := item.(int)
+			i := item
 			result = append(result, i)
 		}
 	}, WithContext(ctx))
@@ -604,19 +587,19 @@ func TestMapReduceWithContext(t *testing.T) {
 func BenchmarkMapReduce(b *testing.B) {
 	b.ReportAllocs()
 
-	mapper := func(v any, writer Writer, cancel func(error)) {
-		writer.Write(v.(int64) * v.(int64))
+	mapper := func(v int64, writer Writer[int64], cancel func(error)) {
+		writer.Write(v * v)
 	}
-	reducer := func(input <-chan any, writer Writer, cancel func(error)) {
+	reducer := func(input <-chan int64, writer Writer[int64], cancel func(error)) {
 		var result int64
 		for v := range input {
-			result += v.(int64)
+			result += v
 		}
 		writer.Write(result)
 	}
 
 	for i := 0; i < b.N; i++ {
-		MapReduce(func(input chan<- any) {
+		MapReduce(func(input chan<- int64) {
 			for j := 0; j < 2; j++ {
 				input <- int64(j)
 			}
diff --git a/core/stores/cache/cache_test.go b/core/stores/cache/cache_test.go
index 83912492..d46e20cd 100644
--- a/core/stores/cache/cache_test.go
+++ b/core/stores/cache/cache_test.go
@@ -163,12 +163,10 @@ func TestCache_SetDel(t *testing.T) {
 		r1, err := miniredis.Run()
 		assert.NoError(t, err)
 		defer r1.Close()
-		r1.SetError("mock error")
 
 		r2, err := miniredis.Run()
 		assert.NoError(t, err)
 		defer r2.Close()
-		r2.SetError("mock error")
 
 		conf := ClusterConf{
 			{
@@ -187,6 +185,8 @@ func TestCache_SetDel(t *testing.T) {
 			},
 		}
 		c := New(conf, syncx.NewSingleFlight(), NewStat("mock"), errPlaceholder)
+		r1.SetError("mock error")
+		r2.SetError("mock error")
 		assert.NoError(t, c.Del("a", "b", "c"))
 	})
 }
diff --git a/core/stores/redis/redis_test.go b/core/stores/redis/redis_test.go
index 2273f713..ec36e02b 100644
--- a/core/stores/redis/redis_test.go
+++ b/core/stores/redis/redis_test.go
@@ -16,6 +16,116 @@ import (
 	"github.com/zeromicro/go-zero/core/stringx"
 )
 
+func TestNewRedis(t *testing.T) {
+	r1, err := miniredis.Run()
+	assert.NoError(t, err)
+	defer r1.Close()
+
+	r2, err := miniredis.Run()
+	assert.NoError(t, err)
+	defer r2.Close()
+	r2.SetError("mock")
+
+	tests := []struct {
+		name string
+		RedisConf
+		ok       bool
+		redisErr bool
+	}{
+		{
+			name: "missing host",
+			RedisConf: RedisConf{
+				Host: "",
+				Type: NodeType,
+				Pass: "",
+			},
+			ok: false,
+		},
+		{
+			name: "missing type",
+			RedisConf: RedisConf{
+				Host: "localhost:6379",
+				Type: "",
+				Pass: "",
+			},
+			ok: false,
+		},
+		{
+			name: "ok",
+			RedisConf: RedisConf{
+				Host: r1.Addr(),
+				Type: NodeType,
+				Pass: "",
+			},
+			ok: true,
+		},
+		{
+			name: "ok",
+			RedisConf: RedisConf{
+				Host: r1.Addr(),
+				Type: ClusterType,
+				Pass: "",
+			},
+			ok: true,
+		},
+		{
+			name: "password",
+			RedisConf: RedisConf{
+				Host: r1.Addr(),
+				Type: NodeType,
+				Pass: "pw",
+			},
+			ok: true,
+		},
+		{
+			name: "tls",
+			RedisConf: RedisConf{
+				Host: r1.Addr(),
+				Type: NodeType,
+				Tls:  true,
+			},
+			ok: true,
+		},
+		{
+			name: "node error",
+			RedisConf: RedisConf{
+				Host: r2.Addr(),
+				Type: NodeType,
+				Pass: "",
+			},
+			ok:       true,
+			redisErr: true,
+		},
+		{
+			name: "cluster error",
+			RedisConf: RedisConf{
+				Host: r2.Addr(),
+				Type: ClusterType,
+				Pass: "",
+			},
+			ok:       true,
+			redisErr: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(stringx.RandId(), func(t *testing.T) {
+			rds, err := NewRedis(test.RedisConf)
+			if test.ok {
+				if test.redisErr {
+					assert.Error(t, err)
+					assert.Nil(t, rds)
+				} else {
+					assert.NoError(t, err)
+					assert.NotNil(t, rds)
+				}
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
 func TestRedis_Decr(t *testing.T) {
 	runOnRedis(t, func(client *Redis) {
 		_, err := New(client.Addr, badType()).Decr("a")
diff --git a/zrpc/server_test.go b/zrpc/server_test.go
index 4cdcb8d8..d35f6c3a 100644
--- a/zrpc/server_test.go
+++ b/zrpc/server_test.go
@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/alicebob/miniredis/v2"
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/discov"
 	"github.com/zeromicro/go-zero/core/logx"
@@ -16,12 +17,16 @@ import (
 )
 
 func TestServer_setupInterceptors(t *testing.T) {
+	rds, err := miniredis.Run()
+	assert.NoError(t, err)
+	defer rds.Close()
+
 	server := new(mockedServer)
-	err := setupInterceptors(server, RpcServerConf{
+	conf := RpcServerConf{
 		Auth: true,
 		Redis: redis.RedisKeyConf{
 			RedisConf: redis.RedisConf{
-				Host: "any",
+				Host: rds.Addr(),
 				Type: redis.NodeType,
 			},
 			Key: "foo",
@@ -35,10 +40,15 @@ func TestServer_setupInterceptors(t *testing.T) {
 			Prometheus: true,
 			Breaker:    true,
 		},
-	}, new(stat.Metrics))
+	}
+	err = setupInterceptors(server, conf, new(stat.Metrics))
 	assert.Nil(t, err)
 	assert.Equal(t, 3, len(server.unaryInterceptors))
 	assert.Equal(t, 1, len(server.streamInterceptors))
+
+	rds.SetError("mock error")
+	err = setupInterceptors(server, conf, new(stat.Metrics))
+	assert.Error(t, err)
 }
 
 func TestServer(t *testing.T) {