Repository: matrixorigin/matrixone
Pull request #4561 (2022-08-18 01:49:53)

Enable flexible `morpc` option for package `logservice`
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #4003

## What this PR does / why we need it:
- Enable flexible `morpc` option for package `logservice`
- Extend `morpc.WithBackendFilter` in order to support network partition feature


Patch:
diff --git a/pkg/common/morpc/backend.go b/pkg/common/morpc/backend.go
index f8de83498..179ca205a 100644
--- a/pkg/common/morpc/backend.go
+++ b/pkg/common/morpc/backend.go
@@ -58,7 +58,7 @@ func WithBackendBusyBufferSize(size int) BackendOption {
 
 // WithBackendFilter set send fiter func. Input ready to send futures, output
 // is really need to be send futures.
-func WithBackendFilter(filter func(Message) bool) BackendOption {
+func WithBackendFilter(filter func(Message, string) bool) BackendOption {
 	return func(rb *remoteBackend) {
 		rb.options.filter = filter
 	}
@@ -120,7 +120,7 @@ type remoteBackend struct {
 		busySize         int
 		batchSendSize    int
 		streamBufferSize int
-		filter           func(Message) bool
+		filter           func(msg Message, backendAddr string) bool
 	}
 
 	stateMu struct {
@@ -217,7 +217,7 @@ func (rb *remoteBackend) adjust() {
 		rb.options.streamBufferSize = 16
 	}
 	if rb.options.filter == nil {
-		rb.options.filter = func(Message) bool {
+		rb.options.filter = func(Message, string) bool {
 			return true
 		}
 	}
@@ -413,7 +413,7 @@ func (rb *remoteBackend) writeLoop(ctx context.Context) {
 				written := 0
 				writeTimeout := time.Duration(0)
 				for _, f := range futures {
-					if rb.options.filter(f.request) && !f.timeout() {
+					if rb.options.filter(f.request, rb.remote) && !f.timeout() {
 						writeTimeout += f.opts.Timeout
 						if err := rb.conn.Write(f.request, goetty.WriteOptions{}); err != nil {
 							rb.logger.Error("write request failed",
@@ -430,7 +430,7 @@ func (rb *remoteBackend) writeLoop(ctx context.Context) {
 				if written > 0 {
 					if err := rb.conn.Flush(writeTimeout); err != nil {
 						for _, f := range futures {
-							if rb.options.filter(f.request) {
+							if rb.options.filter(f.request, rb.remote) {
 								rb.logger.Error("write request failed",
 									zap.Uint64("request-id", f.request.GetID()),
 									zap.Error(err))
diff --git a/pkg/dnservice/store.go b/pkg/dnservice/store.go
index f583c7ec8..a94aaa02f 100644
--- a/pkg/dnservice/store.go
+++ b/pkg/dnservice/store.go
@@ -51,10 +51,10 @@ func WithConfigAdjust(adjustConfigFunc func(c *Config)) Option {
 	}
 }
 
-// WithRequestFilter set filtering txn.TxnRequest sent to other DNShard
-func WithRequestFilter(filter func(*txn.TxnRequest) bool) Option {
+// WithBackendFilter set filtering txn.TxnRequest sent to other DNShard
+func WithBackendFilter(filter func(*txn.TxnRequest, string) bool) Option {
 	return func(s *store) {
-		s.options.requestFilter = filter
+		s.options.backendFilter = filter
 	}
 }
 
@@ -88,7 +88,7 @@ type store struct {
 	options struct {
 		logServiceClientFactory func(metadata.DNShard) (logservice.Client, error)
 		hakeekerClientFactory   func() (logservice.DNHAKeeperClient, error)
-		requestFilter           func(*txn.TxnRequest) bool
+		backendFilter           func(req *txn.TxnRequest, backendAddr string) bool
 		adjustConfigFunc        func(c *Config)
 	}
 
@@ -390,8 +390,8 @@ func (s *store) initFileService() error {
 func (s *store) getBackendOptions() []morpc.BackendOption {
 	return []morpc.BackendOption{
 		morpc.WithBackendLogger(s.logger),
-		morpc.WithBackendFilter(func(m morpc.Message) bool {
-			return s.options.requestFilter == nil || s.options.requestFilter(m.(*txn.TxnRequest))
+		morpc.WithBackendFilter(func(m morpc.Message, backendAddr string) bool {
+			return s.options.backendFilter == nil || s.options.backendFilter(m.(*txn.TxnRequest), backendAddr)
 		}),
 		morpc.WithBackendBusyBufferSize(s.cfg.RPC.BusyQueueSize),
 		morpc.WithBackendBufferSize(s.cfg.RPC.SendQueueSize),
diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go
index bf396930e..11ef02f70 100644
--- a/pkg/logservice/client.go
+++ b/pkg/logservice/client.go
@@ -511,16 +511,27 @@ func getRPCClient(ctx context.Context, target string, pool *sync.Pool) (morpc.RP
 	mf := func() morpc.Message {
 		return pool.Get().(*RPCResponse)
 	}
+
+	// construct morpc.BackendOption
+	backendOpts := []morpc.BackendOption{
+		morpc.WithBackendConnectWhenCreate(),
+		morpc.WithBackendConnectTimeout(time.Second),
+	}
+	backendOpts = append(backendOpts, GetBackendOptions(ctx)...)
+
+	// construct morpc.ClientOption
+	clientOpts := []morpc.ClientOption{
+		morpc.WithClientInitBackends([]string{target}, []int{1}),
+		morpc.WithClientMaxBackendPerHost(1),
+	}
+	clientOpts = append(clientOpts, GetClientOptions(ctx)...)
+
 	// we set connection timeout to a constant value so if ctx's deadline is much
 	// larger, then we can ensure that all specified potential nodes have a chance
 	// to be attempted
 	codec := morpc.NewMessageCodec(mf, defaultWriteSocketSize)
-	bf := morpc.NewGoettyBasedBackendFactory(codec,
-		morpc.WithBackendConnectWhenCreate(),
-		morpc.WithBackendConnectTimeout(time.Second))
-	return morpc.NewClient(bf,
-		morpc.WithClientInitBackends([]string{target}, []int{1}),
-		morpc.WithClientMaxBackendPerHost(1))
+	bf := morpc.NewGoettyBasedBackendFactory(codec, backendOpts...)
+	return morpc.NewClient(bf, clientOpts...)
 }
 
 func getTimeoutFromContext(ctx context.Context) (time.Duration, error) {
diff --git a/pkg/logservice/option.go b/pkg/logservice/option.go
new file mode 100644
index 000000000..f9bc58da9
--- /dev/null
+++ b/pkg/logservice/option.go
@@ -0,0 +1,60 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"context"
+
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
+)
+
+// Option is utility that sets callback for Service.
+type Option func(*Service)
+
+// WithBackendFilter sets filter via which could select remote backend.
+func WithBackendFilter(filter func(morpc.Message, string) bool) Option {
+	return func(s *Service) {
+		s.options.backendFilter = filter
+	}
+}
+
+type ContextKey string
+
+const (
+	BackendOption ContextKey = "morpc.BackendOption"
+	ClientOption  ContextKey = "morpc.ClientOption"
+)
+
+func GetBackendOptions(ctx context.Context) []morpc.BackendOption {
+	if v := ctx.Value(BackendOption); v != nil {
+		return v.([]morpc.BackendOption)
+	}
+	return nil
+}
+
+func GetClientOptions(ctx context.Context) []morpc.ClientOption {
+	if v := ctx.Value(ClientOption); v != nil {
+		return v.([]morpc.ClientOption)
+	}
+	return nil
+}
+
+func SetBackendOptions(ctx context.Context, opts ...morpc.BackendOption) context.Context {
+	return context.WithValue(ctx, BackendOption, opts)
+}
+
+func SetClientOptions(ctx context.Context, opts ...morpc.ClientOption) context.Context {
+	return context.WithValue(ctx, ClientOption, opts)
+}
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index 30d28f10d..fc57a72e7 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -65,9 +65,14 @@ type Service struct {
 	respPool *sync.Pool
 	stopper  *stopper.Stopper
 	haClient LogHAKeeperClient
+
+	options struct {
+		// morpc client would filter remote backend via this
+		backendFilter func(msg morpc.Message, backendAddr string) bool
+	}
 }
 
-func NewService(cfg Config) (*Service, error) {
+func NewService(cfg Config, opts ...Option) (*Service, error) {
 	cfg.Fill()
 	if err := cfg.Validate(); err != nil {
 		return nil, err
@@ -111,6 +116,9 @@ func NewService(cfg Config) (*Service, error) {
 		respPool: respPool,
 		stopper:  stopper.NewStopper("log-service"),
 	}
+	for _, opt := range opts {
+		opt(service)
+	}
 	server.RegisterRequestHandler(service.handleRPCRequest)
 	// TODO: before making the service available to the outside world, restore all
 	// replicas already known to the local store
@@ -125,6 +133,10 @@ func NewService(cfg Config) (*Service, error) {
 	if !cfg.DisableWorkers {
 		if err := service.stopper.RunNamedTask("log-heartbeat-worker", func(ctx context.Context) {
 			plog.Infof("logservice heartbeat worker started")
+
+			// transfer morpc options via context
+			ctx = SetBackendOptions(ctx, service.getBackendOptions()...)
+			ctx = SetClientOptions(ctx, service.getClientOptions()...)
 			service.heartbeatWorker(ctx)
 		}); err != nil {
 			return nil, err
@@ -373,3 +385,17 @@ func (s *Service) handleCheckHAKeeper(req pb.Request) pb.Response {
 	}
 	return resp
 }
+
+func (s *Service) getBackendOptions() []morpc.BackendOption {
+	return []morpc.BackendOption{
+		morpc.WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return s.options.backendFilter == nil ||
+				s.options.backendFilter(msg.(*RPCRequest), backendAddr)
+		}),
+	}
+}
+
+// NB: leave an empty method for future extension.
+func (s *Service) getClientOptions() []morpc.ClientOption {
+	return nil
+}
diff --git a/pkg/logservice/service_wrap.go b/pkg/logservice/service_wrap.go
index 1df6a8b4a..ef0f1cdd9 100644
--- a/pkg/logservice/service_wrap.go
+++ b/pkg/logservice/service_wrap.go
@@ -23,8 +23,8 @@ type WrappedService struct {
 	svc *Service
 }
 
-func NewWrappedService(c Config) (*WrappedService, error) {
-	svc, err := NewService(c)
+func NewWrappedService(c Config, opts ...Option) (*WrappedService, error) {
+	svc, err := NewService(c, opts...)
 	if err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/pkg/common/morpc/backend_test.go b/pkg/common/morpc/backend_test.go
index 1f292b83e..4b96cdf01 100644
--- a/pkg/common/morpc/backend_test.go
+++ b/pkg/common/morpc/backend_test.go
@@ -125,7 +125,7 @@ func TestSendWithAlreadyContextDone(t *testing.T) {
 			assert.Nil(t, resp)
 		},
 		WithBackendConnectWhenCreate(),
-		WithBackendFilter(func(Message) bool {
+		WithBackendFilter(func(Message, string) bool {
 			cancel()
 			return true
 		}))
@@ -150,7 +150,7 @@ func TestSendWithResetConnAndRetry(t *testing.T) {
 			assert.Equal(t, req, resp)
 			assert.True(t, retry > 0)
 		},
-		WithBackendFilter(func(Message) bool {
+		WithBackendFilter(func(Message, string) bool {
 			retry++
 			return true
 		}))
@@ -201,7 +201,7 @@ func TestSendWithReconnect(t *testing.T) {
 			}
 		},
 		WithBackendConnectWhenCreate(),
-		WithBackendFilter(func(m Message) bool {
+		WithBackendFilter(func(Message, string) bool {
 			idx++
 			if idx%2 == 0 {
 				rb.closeConn(false)
@@ -231,7 +231,7 @@ func TestSendWithCannotConnectWillTimeout(t *testing.T) {
 			assert.Nil(t, resp)
 			assert.Equal(t, err, ctx.Err())
 		},
-		WithBackendFilter(func(m Message) bool {
+		WithBackendFilter(func(Message, string) bool {
 			assert.NoError(t, rb.conn.Disconnect())
 			rb.remote = ""
 			return true
@@ -319,7 +319,7 @@ func TestBusy(t *testing.T) {
 			c <- struct{}{}
 		},
 		WithBackendConnectWhenCreate(),
-		WithBackendFilter(func(Message) bool {
+		WithBackendFilter(func(Message, string) bool {
 			if n == 0 {
 				<-c
 				n++
diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go
index ba921594e..a43a01798 100644
--- a/pkg/logservice/client_test.go
+++ b/pkg/logservice/client_test.go
@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -36,7 +37,11 @@ func runClientTest(t *testing.T,
 	defer leaktest.AfterTest(t)()
 	cfg := getServiceTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	service, err := NewService(cfg)
+	service, err := NewService(cfg,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service.Close())
@@ -119,7 +124,11 @@ func TestClientCanBeConnectedByReverseProxy(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getServiceTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	service, err := NewService(cfg)
+	service, err := NewService(cfg,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service.Close())
diff --git a/pkg/logservice/hakeeper_client_test.go b/pkg/logservice/hakeeper_client_test.go
index a2f524c28..8d51e91e9 100644
--- a/pkg/logservice/hakeeper_client_test.go
+++ b/pkg/logservice/hakeeper_client_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
@@ -303,13 +304,21 @@ func testNotHAKeeperErrorIsHandled(t *testing.T, fn func(*testing.T, *managedHAK
 		DisableWorkers:      true,
 	}
 	cfg1.Fill()
-	service1, err := NewService(cfg1)
+	service1, err := NewService(cfg1,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service1.Close())
 	}()
 	cfg2.Fill()
-	service2, err := NewService(cfg2)
+	service2, err := NewService(cfg2,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service2.Close())
diff --git a/pkg/logservice/option_test.go b/pkg/logservice/option_test.go
new file mode 100644
index 000000000..f530314e2
--- /dev/null
+++ b/pkg/logservice/option_test.go
@@ -0,0 +1,61 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logservice
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
+)
+
+func TestWithOption(t *testing.T) {
+	ctx := context.Background()
+	require.Nil(t, GetBackendOptions(ctx))
+	require.Nil(t, GetClientOptions(ctx))
+
+	// set morpc.BackendOption
+	backendOpts := []morpc.BackendOption{
+		morpc.WithBackendConnectWhenCreate(),
+		morpc.WithBackendConnectTimeout(time.Second),
+	}
+	ctx1 := SetBackendOptions(ctx, backendOpts...)
+	backendOptsRet := GetBackendOptions(ctx1)
+	require.NotNil(t, backendOptsRet)
+	require.Equal(t, len(backendOpts), len(backendOptsRet))
+	require.Equal(t, backendOpts, backendOptsRet)
+
+	// set morpc.BackendOption again
+	newBackendOpts := []morpc.BackendOption{
+		morpc.WithBackendConnectWhenCreate(),
+	}
+	ctx2 := SetBackendOptions(ctx1, newBackendOpts...)
+	newBackendOptsRet := GetBackendOptions(ctx2)
+	require.NotNil(t, backendOptsRet)
+	require.Equal(t, len(newBackendOpts), len(newBackendOptsRet))
+	require.Equal(t, newBackendOpts, newBackendOptsRet)
+
+	clientOpts := []morpc.ClientOption{
+		morpc.WithClientMaxBackendPerHost(1),
+	}
+	ctx3 := SetClientOptions(ctx, clientOpts...)
+	clientOptsRet := GetClientOptions(ctx3)
+	require.NotNil(t, clientOptsRet)
+	require.Equal(t, len(clientOpts), len(clientOptsRet))
+	require.Equal(t, clientOpts, clientOptsRet)
+}
diff --git a/pkg/logservice/service_commands_test.go b/pkg/logservice/service_commands_test.go
index 070bed20a..e7c13110d 100644
--- a/pkg/logservice/service_commands_test.go
+++ b/pkg/logservice/service_commands_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -46,7 +47,11 @@ func TestBackgroundTickAndHeartbeat(t *testing.T) {
 	cfg.HAKeeperTickInterval.Duration = 5 * time.Millisecond
 	cfg.HAKeeperClientConfig.ServiceAddresses = []string{"127.0.0.1:9002"}
 	cfg.Fill()
-	service, err := NewService(cfg)
+	service, err := NewService(cfg,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service.Close())
diff --git a/pkg/logservice/service_test.go b/pkg/logservice/service_test.go
index c5f52dcc0..f71b42499 100644
--- a/pkg/logservice/service_test.go
+++ b/pkg/logservice/service_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -56,7 +57,11 @@ func runServiceTest(t *testing.T,
 	defer leaktest.AfterTest(t)()
 	cfg := getServiceTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	service, err := NewService(cfg)
+	service, err := NewService(cfg,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	peers := make(map[uint64]dragonboat.Target)
 	peers[1] = service.ID()
@@ -79,7 +84,11 @@ func TestNewService(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	cfg := getServiceTestConfig()
 	defer vfs.ReportLeakedFD(cfg.FS, t)
-	service, err := NewService(cfg)
+	service, err := NewService(cfg,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	assert.NoError(t, service.Close())
 }
@@ -508,7 +517,11 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 		DisableWorkers:      true,
 	}
 	cfg1.Fill()
-	service1, err := NewService(cfg1)
+	service1, err := NewService(cfg1,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service1.Close())
@@ -517,7 +530,11 @@ func TestShardInfoCanBeQueried(t *testing.T) {
 	peers1[1] = service1.ID()
 	assert.NoError(t, service1.store.startReplica(1, 1, peers1, false))
 	cfg2.Fill()
-	service2, err := NewService(cfg2)
+	service2, err := NewService(cfg2,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service2.Close())
@@ -629,7 +646,11 @@ func TestGossipInSimulatedCluster(t *testing.T) {
 		}
 		cfg.GossipProbeInterval.Duration = 350 * time.Millisecond
 		configs = append(configs, cfg)
-		service, err := NewService(cfg)
+		service, err := NewService(cfg,
+			WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+				return true
+			}),
+		)
 		require.NoError(t, err)
 		services = append(services, service)
 	}
@@ -732,7 +753,11 @@ func TestGossipInSimulatedCluster(t *testing.T) {
 	require.NoError(t, services[12].Close())
 	services[12] = nil
 	time.Sleep(2 * time.Second)
-	service, err := NewService(configs[12])
+	service, err := NewService(configs[12],
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		require.NoError(t, service.Close())
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index d50e016cb..68c62355a 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -26,6 +26,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
@@ -194,25 +195,41 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	cfg4.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
 	cfg4.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
 	cfg1.Fill()
-	service1, err := NewService(cfg1)
+	service1, err := NewService(cfg1,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service1.Close())
 	}()
 	cfg2.Fill()
-	service2, err := NewService(cfg2)
+	service2, err := NewService(cfg2,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service2.Close())
 	}()
 	cfg3.Fill()
-	service3, err := NewService(cfg3)
+	service3, err := NewService(cfg3,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service3.Close())
 	}()
 	cfg4.Fill()
-	service4, err := NewService(cfg4)
+	service4, err := NewService(cfg4,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	require.NoError(t, err)
 	defer func() {
 		assert.NoError(t, service4.Close())
diff --git a/pkg/logservice/testclient.go b/pkg/logservice/testclient.go
index 47ac34a27..e5bca86b7 100644
--- a/pkg/logservice/testclient.go
+++ b/pkg/logservice/testclient.go
@@ -19,6 +19,7 @@ import (
 
 	"github.com/google/uuid"
 
+	"github.com/matrixorigin/matrixone/pkg/common/morpc"
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
@@ -35,7 +36,11 @@ func NewTestService(fs vfs.FS) (*Service, ClientConfig, error) {
 		DisableWorkers:       true,
 	}
 	cfg.Fill()
-	service, err := NewService(cfg)
+	service, err := NewService(cfg,
+		WithBackendFilter(func(msg morpc.Message, backendAddr string) bool {
+			return true
+		}),
+	)
 	if err != nil {
 		return nil, ClientConfig{}, err
 	}