Repository: matrixorigin/matrixone
Pull request #4461 (2022-08-11 07:39:31)

Implement interface `ClusterWaitState` for test framework;
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] 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:
- Implement interface `ClusterWaitState` which is used to wait for some expected state;
- Add more convenient methods within interface `ClusterState`;
- Add unit test, and make all unit tests runnable;


Patch:
diff --git a/pkg/hakeeper/config.go b/pkg/hakeeper/config.go
index 31d7c4951..4a0212391 100644
--- a/pkg/hakeeper/config.go
+++ b/pkg/hakeeper/config.go
@@ -21,7 +21,7 @@ import (
 const (
 	DefaultTickPerSecond   = 10
 	DefaultLogStoreTimeout = 5 * time.Minute
-	DefaultDnStoreTimeout  = 10 * time.Second
+	DefaultDNStoreTimeout  = 10 * time.Second
 )
 
 type Config struct {
@@ -35,10 +35,10 @@ type Config struct {
 	// it regards the log store as down.
 	LogStoreTimeout time.Duration
 
-	// DnStoreTimeout is the actual time limit between a dn store's heartbeat.
-	// If HAKeeper does not receive two heartbeat within DnStoreTimeout,
+	// DNStoreTimeout is the actual time limit between a dn store's heartbeat.
+	// If HAKeeper does not receive two heartbeat within DNStoreTimeout,
 	// it regards the dn store as down.
-	DnStoreTimeout time.Duration
+	DNStoreTimeout time.Duration
 }
 
 func (cfg Config) Validate() error {
@@ -52,8 +52,8 @@ func (cfg *Config) Fill() {
 	if cfg.LogStoreTimeout == 0 {
 		cfg.LogStoreTimeout = DefaultLogStoreTimeout
 	}
-	if cfg.DnStoreTimeout == 0 {
-		cfg.DnStoreTimeout = DefaultDnStoreTimeout
+	if cfg.DNStoreTimeout == 0 {
+		cfg.DNStoreTimeout = DefaultDNStoreTimeout
 	}
 }
 
@@ -62,7 +62,7 @@ func (cfg Config) LogStoreExpired(start, current uint64) bool {
 }
 
 func (cfg Config) DnStoreExpired(start, current uint64) bool {
-	return uint64(int(cfg.DnStoreTimeout/time.Second)*cfg.TickPerSecond)+start < current
+	return uint64(int(cfg.DNStoreTimeout/time.Second)*cfg.TickPerSecond)+start < current
 }
 
 func (cfg Config) ExpiredTick(start uint64, timeout time.Duration) uint64 {
diff --git a/pkg/logservice/config.go b/pkg/logservice/config.go
index d6f18b39e..36e37f7bc 100644
--- a/pkg/logservice/config.go
+++ b/pkg/logservice/config.go
@@ -133,10 +133,10 @@ type Config struct {
 		// If HAKeeper does not receive two heartbeat within LogStoreTimeout,
 		// it regards the log store as down.
 		LogStoreTimeout toml.Duration `toml:"log-store-timeout"`
-		// DnStoreTimeout is the actual time limit between a dn store's heartbeat.
-		// If HAKeeper does not receive two heartbeat within DnStoreTimeout,
+		// DNStoreTimeout is the actual time limit between a dn store's heartbeat.
+		// If HAKeeper does not receive two heartbeat within DNStoreTimeout,
 		// it regards the dn store as down.
-		DnStoreTimeout toml.Duration `toml:"dn-store-timeout"`
+		DNStoreTimeout toml.Duration `toml:"dn-store-timeout"`
 	}
 
 	// HAKeeperClientConfig is the config for HAKeeperClient
@@ -150,7 +150,7 @@ func (c *Config) GetHAKeeperConfig() hakeeper.Config {
 	return hakeeper.Config{
 		TickPerSecond:   c.HAKeeperConfig.TickPerSecond,
 		LogStoreTimeout: c.HAKeeperConfig.LogStoreTimeout.Duration,
-		DnStoreTimeout:  c.HAKeeperConfig.DnStoreTimeout.Duration,
+		DNStoreTimeout:  c.HAKeeperConfig.DNStoreTimeout.Duration,
 	}
 }
 
@@ -239,8 +239,8 @@ func (c *Config) Validate() error {
 	if c.HAKeeperConfig.LogStoreTimeout.Duration == 0 {
 		return errors.Wrapf(ErrInvalidConfig, "LogStoreTimeout not set")
 	}
-	if c.HAKeeperConfig.DnStoreTimeout.Duration == 0 {
-		return errors.Wrapf(ErrInvalidConfig, "DnStoreTimeout not set")
+	if c.HAKeeperConfig.DNStoreTimeout.Duration == 0 {
+		return errors.Wrapf(ErrInvalidConfig, "DNStoreTimeout not set")
 	}
 	// validate BootstrapConfig
 	if c.BootstrapConfig.BootstrapCluster {
@@ -305,8 +305,8 @@ func (c *Config) Fill() {
 	if c.HAKeeperConfig.LogStoreTimeout.Duration == 0 {
 		c.HAKeeperConfig.LogStoreTimeout.Duration = hakeeper.DefaultLogStoreTimeout
 	}
-	if c.HAKeeperConfig.DnStoreTimeout.Duration == 0 {
-		c.HAKeeperConfig.DnStoreTimeout.Duration = hakeeper.DefaultDnStoreTimeout
+	if c.HAKeeperConfig.DNStoreTimeout.Duration == 0 {
+		c.HAKeeperConfig.DNStoreTimeout.Duration = hakeeper.DefaultDNStoreTimeout
 	}
 	if c.HeartbeatInterval.Duration == 0 {
 		c.HeartbeatInterval.Duration = defaultHeartbeatInterval
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 54e885bbe..18aefcbd6 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -132,8 +132,8 @@ func newLogStore(cfg Config) (*store, error) {
 		return nil, err
 	}
 	hakeeperConfig := cfg.GetHAKeeperConfig()
-	plog.Infof("HAKeeper LogStoreTimeout: %s, DnStoreTimeout: %s",
-		hakeeperConfig.LogStoreTimeout, hakeeperConfig.DnStoreTimeout)
+	plog.Infof("HAKeeper LogStoreTimeout: %s, DNStoreTimeout: %s",
+		hakeeperConfig.LogStoreTimeout, hakeeperConfig.DNStoreTimeout)
 	ls := &store{
 		cfg:           cfg,
 		nh:            nh,


Test patch:
diff --git a/pkg/hakeeper/checkers/dnservice/check_test.go b/pkg/hakeeper/checkers/dnservice/check_test.go
index ca1ae77a7..9f961ce48 100644
--- a/pkg/hakeeper/checkers/dnservice/check_test.go
+++ b/pkg/hakeeper/checkers/dnservice/check_test.go
@@ -194,7 +194,7 @@ func TestCheck(t *testing.T) {
 	// construct current tick in order to make heartbeat tick expired
 	config := hakeeper.Config{}
 	config.Fill()
-	currTick := config.ExpiredTick(staleTick, config.DnStoreTimeout) + 1
+	currTick := config.ExpiredTick(staleTick, config.DNStoreTimeout) + 1
 
 	enough := true
 	newReplicaID := uint64(100)
diff --git a/pkg/hakeeper/checkers/dnservice/parse_test.go b/pkg/hakeeper/checkers/dnservice/parse_test.go
index 5933f2950..a6a151225 100644
--- a/pkg/hakeeper/checkers/dnservice/parse_test.go
+++ b/pkg/hakeeper/checkers/dnservice/parse_test.go
@@ -56,14 +56,14 @@ func TestCheckInitatingShards(t *testing.T) {
 	config := hakeeper.Config{
 		TickPerSecond:   10,
 		LogStoreTimeout: 10 * time.Second,
-		DnStoreTimeout:  10 * time.Second,
+		DNStoreTimeout:  10 * time.Second,
 	}
 
 	// mock ShardMapper
 	mapper := mockShardMapper()
 
 	earliestTick := uint64(10)
-	expiredTick := config.ExpiredTick(earliestTick, config.DnStoreTimeout) + 1
+	expiredTick := config.ExpiredTick(earliestTick, config.DNStoreTimeout) + 1
 
 	// discover an initial shard => no operators generated
 	ops := checkInitatingShards(rs, mapper, workingStores, idAlloc, cluster, config, earliestTick)
@@ -149,7 +149,7 @@ func TestParseDNState(t *testing.T) {
 	// construct current tick in order to make heartbeat tick expired
 	cfg := hakeeper.Config{}
 	cfg.Fill()
-	currTick := cfg.ExpiredTick(expiredTick, cfg.DnStoreTimeout) + 1
+	currTick := cfg.ExpiredTick(expiredTick, cfg.DNStoreTimeout) + 1
 
 	// 1. no working dn stores
 	{
diff --git a/pkg/hakeeper/checkers/syshealth/check_test.go b/pkg/hakeeper/checkers/syshealth/check_test.go
index fd5908cea..b34d093b0 100644
--- a/pkg/hakeeper/checkers/syshealth/check_test.go
+++ b/pkg/hakeeper/checkers/syshealth/check_test.go
@@ -104,7 +104,7 @@ func TestParseDnStores(t *testing.T) {
 	// construct current tick in order to make heartbeat tick expired
 	cfg := hakeeper.Config{}
 	cfg.Fill()
-	currTick := cfg.ExpiredTick(expiredTick, cfg.DnStoreTimeout) + 1
+	currTick := cfg.ExpiredTick(expiredTick, cfg.DNStoreTimeout) + 1
 
 	dnState := pb.DNState{
 		Stores: map[string]pb.DNStoreInfo{
diff --git a/pkg/hakeeper/config_test.go b/pkg/hakeeper/config_test.go
index e080897ef..408d001f3 100644
--- a/pkg/hakeeper/config_test.go
+++ b/pkg/hakeeper/config_test.go
@@ -15,8 +15,9 @@
 package hakeeper
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestTimeoutConfig(t *testing.T) {
@@ -24,5 +25,5 @@ func TestTimeoutConfig(t *testing.T) {
 	c.Fill()
 	assert.Equal(t, DefaultTickPerSecond, c.TickPerSecond)
 	assert.Equal(t, DefaultLogStoreTimeout, c.LogStoreTimeout)
-	assert.Equal(t, DefaultDnStoreTimeout, c.DnStoreTimeout)
+	assert.Equal(t, DefaultDNStoreTimeout, c.DNStoreTimeout)
 }
diff --git a/pkg/logservice/config_test.go b/pkg/logservice/config_test.go
index 3d86635b6..f69f172a6 100644
--- a/pkg/logservice/config_test.go
+++ b/pkg/logservice/config_test.go
@@ -170,7 +170,7 @@ func TestFillConfig(t *testing.T) {
 	assert.Equal(t, 0, len(c.GossipSeedAddresses))
 	assert.Equal(t, hakeeper.DefaultTickPerSecond, c.HAKeeperConfig.TickPerSecond)
 	assert.Equal(t, hakeeper.DefaultLogStoreTimeout, c.HAKeeperConfig.LogStoreTimeout.Duration)
-	assert.Equal(t, hakeeper.DefaultDnStoreTimeout, c.HAKeeperConfig.DnStoreTimeout.Duration)
+	assert.Equal(t, hakeeper.DefaultDNStoreTimeout, c.HAKeeperConfig.DNStoreTimeout.Duration)
 }
 
 func TestListenAddressCanBeFilled(t *testing.T) {
diff --git a/pkg/logservice/store_hakeeper_check_test.go b/pkg/logservice/store_hakeeper_check_test.go
index ae83d26a6..d50e016cb 100644
--- a/pkg/logservice/store_hakeeper_check_test.go
+++ b/pkg/logservice/store_hakeeper_check_test.go
@@ -147,7 +147,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	}
 	cfg1.HAKeeperConfig.TickPerSecond = 10
 	cfg1.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
-	cfg1.HAKeeperConfig.DnStoreTimeout.Duration = 10 * time.Second
+	cfg1.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
 	cfg2 := Config{
 		UUID:                uuid.New().String(),
 		FS:                  vfs.NewStrictMem(),
@@ -162,7 +162,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	}
 	cfg2.HAKeeperConfig.TickPerSecond = 10
 	cfg2.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
-	cfg2.HAKeeperConfig.DnStoreTimeout.Duration = 10 * time.Second
+	cfg2.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
 	cfg3 := Config{
 		UUID:                uuid.New().String(),
 		FS:                  vfs.NewStrictMem(),
@@ -177,7 +177,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	}
 	cfg3.HAKeeperConfig.TickPerSecond = 10
 	cfg3.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
-	cfg3.HAKeeperConfig.DnStoreTimeout.Duration = 10 * time.Second
+	cfg3.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
 	cfg4 := Config{
 		UUID:                uuid.New().String(),
 		FS:                  vfs.NewStrictMem(),
@@ -192,7 +192,7 @@ func runHAKeeperClusterTest(t *testing.T, fn func(*testing.T, []*Service)) {
 	}
 	cfg4.HAKeeperConfig.TickPerSecond = 10
 	cfg4.HAKeeperConfig.LogStoreTimeout.Duration = 5 * time.Second
-	cfg4.HAKeeperConfig.DnStoreTimeout.Duration = 10 * time.Second
+	cfg4.HAKeeperConfig.DNStoreTimeout.Duration = 10 * time.Second
 	cfg1.Fill()
 	service1, err := NewService(cfg1)
 	require.NoError(t, err)
diff --git a/pkg/tests/service/errors.go b/pkg/tests/service/errors.go
index 23e44fb78..9799bf93a 100644
--- a/pkg/tests/service/errors.go
+++ b/pkg/tests/service/errors.go
@@ -14,14 +14,18 @@
 
 package service
 
-import "fmt"
+import (
+	"errors"
+	"fmt"
+)
 
 var (
-	ErrServiceNotExist     = fmt.Errorf("service not exist")
-	ErrServiceNotStarted   = fmt.Errorf("service not started")
-	ErrInvalidServiceIndex = fmt.Errorf("invalid service index")
-	ErrFailAllocatePort    = fmt.Errorf("fail to allocate port")
-	ErrInvalidFSName       = fmt.Errorf("invalid file service name")
+	ErrServiceNotExist     = errors.New("service not exist")
+	ErrServiceNotStarted   = errors.New("service not started")
+	ErrStoreNotReported    = errors.New("store not reported")
+	ErrInvalidServiceIndex = errors.New("invalid service index")
+	ErrFailAllocatePort    = errors.New("fail to allocate port")
+	ErrInvalidFSName       = errors.New("invalid file service name")
 )
 
 // wrappedError wraps error with extra message.
diff --git a/pkg/tests/service/logservice.go b/pkg/tests/service/logservice.go
index d5f1b28bf..9c26049b4 100644
--- a/pkg/tests/service/logservice.go
+++ b/pkg/tests/service/logservice.go
@@ -159,7 +159,7 @@ func buildLogConfig(index int, opt Options, address serviceAddress) logservice.C
 	// setting hakeeper configuration
 	cfg.HAKeeperConfig.TickPerSecond = opt.hakeeper.tickPerSecond
 	cfg.HAKeeperConfig.LogStoreTimeout.Duration = opt.hakeeper.logStoreTimeout
-	cfg.HAKeeperConfig.DnStoreTimeout.Duration = opt.hakeeper.dnStoreTimeout
+	cfg.HAKeeperConfig.DNStoreTimeout.Duration = opt.hakeeper.dnStoreTimeout
 
 	// we must invoke Fill in order to setting default configruation value.
 	cfg.Fill()
diff --git a/pkg/tests/service/options.go b/pkg/tests/service/options.go
index 8fcd6f4c4..1b7a61b27 100644
--- a/pkg/tests/service/options.go
+++ b/pkg/tests/service/options.go
@@ -18,6 +18,8 @@ import (
 	"time"
 
 	"go.uber.org/zap/zapcore"
+
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
 )
 
 const (
@@ -45,7 +47,7 @@ const (
 	// default hakeeper configuration
 	defaultTickPerSecond   = 10
 	defaultLogStoreTimeout = 10 * time.Second
-	defaultDnStoreTimeout  = 10 * time.Second
+	defaultDNStoreTimeout  = 10 * time.Second
 	defaultCheckInterval   = 3 * time.Second
 
 	// default heartbeat configuration
@@ -128,7 +130,7 @@ func (opt *Options) validate() {
 		opt.hakeeper.logStoreTimeout = defaultLogStoreTimeout
 	}
 	if opt.hakeeper.dnStoreTimeout == 0 {
-		opt.hakeeper.dnStoreTimeout = defaultDnStoreTimeout
+		opt.hakeeper.dnStoreTimeout = defaultDNStoreTimeout
 	}
 	if opt.hakeeper.checkInterval == 0 {
 		opt.hakeeper.checkInterval = defaultCheckInterval
@@ -143,6 +145,17 @@ func (opt *Options) validate() {
 	}
 }
 
+// BuildHAKeeperConfig returns hakeeper.Config
+//
+// We could check timeout for dn/log store via hakeeper.Config.
+func (opt Options) BuildHAKeeperConfig() hakeeper.Config {
+	return hakeeper.Config{
+		TickPerSecond:   opt.hakeeper.tickPerSecond,
+		LogStoreTimeout: opt.hakeeper.logStoreTimeout,
+		DNStoreTimeout:  opt.hakeeper.dnStoreTimeout,
+	}
+}
+
 // WithDNServiceNum sets dn service number in the cluster.
 func (opt Options) WithDNServiceNum(num int) Options {
 	opt.initial.dnServiceNum = num
diff --git a/pkg/tests/service/options_test.go b/pkg/tests/service/options_test.go
index 0501f8a9e..1c6e77f96 100644
--- a/pkg/tests/service/options_test.go
+++ b/pkg/tests/service/options_test.go
@@ -101,7 +101,7 @@ func TestWithHKLogStoreTimeout(t *testing.T) {
 
 func TestWithHKDNStoreTimeout(t *testing.T) {
 	opt := DefaultOptions()
-	require.Equal(t, defaultDnStoreTimeout, opt.hakeeper.dnStoreTimeout)
+	require.Equal(t, defaultDNStoreTimeout, opt.hakeeper.dnStoreTimeout)
 
 	timeout := 21 * time.Second
 	opt = opt.WithHKDNStoreTimeout(timeout)
@@ -148,3 +148,11 @@ func TestHaKeeperNum(t *testing.T) {
 	require.Equal(t, 3, haKeeperNum(3))
 	require.Equal(t, 3, haKeeperNum(4))
 }
+
+func TestBuildHAKeeperConfig(t *testing.T) {
+	opt := DefaultOptions()
+	cfg := opt.BuildHAKeeperConfig()
+	require.Equal(t, opt.hakeeper.tickPerSecond, cfg.TickPerSecond)
+	require.Equal(t, opt.hakeeper.logStoreTimeout, cfg.LogStoreTimeout)
+	require.Equal(t, opt.hakeeper.dnStoreTimeout, cfg.DNStoreTimeout)
+}
diff --git a/pkg/tests/service/service.go b/pkg/tests/service/service.go
index 0b4fc7780..09ef43449 100644
--- a/pkg/tests/service/service.go
+++ b/pkg/tests/service/service.go
@@ -28,12 +28,18 @@ import (
 
 	"github.com/matrixorigin/matrixone/pkg/dnservice"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	"github.com/matrixorigin/matrixone/pkg/hakeeper/checkers/syshealth"
 	"github.com/matrixorigin/matrixone/pkg/logservice"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	logpb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 )
 
+var (
+	defaultWaitInterval = 100 * time.Millisecond
+)
+
 // Cluster describes behavior of test framwork.
 type Cluster interface {
 	// Start starts svcs sequentially
@@ -116,15 +122,37 @@ type ClusterState interface {
 
 	// GetHAKeeperState returns hakeeper state from running hakeeper.
 	GetHAKeeperState() logpb.HAKeeperState
+	// GetHAKeeperConfig returns hakeeper configuration.
+	GetHAKeeperConfig() hakeeper.Config
+
+	// DNStoreExpired checks dn store expired or not by uuid.
+	DNStoreExpired(uuid string) (bool, error)
+	// DNStoreExpiredIndexed checks dn store expired or not by index.
+	DNStoreExpiredIndexed(index int) (bool, error)
+	// LogStoreExpired checks log store expired or not by uuid.
+	LogStoreExpired(uuid string) (bool, error)
+	// LogStoreExpiredIndexed checks log store expired or not by index.
+	LogStoreExpiredIndexed(index int) (bool, error)
+
+	// IsClusterHealthy checks whether cluster is healthy or not.
+	IsClusterHealthy() bool
 }
 
 // ClusterWaitState waits cluster state until timeout.
 type ClusterWaitState interface {
+	// WaitHAKeeperLeader waits hakeeper leader elected and return it.
 	WaitHAKeeperLeader(ctx context.Context) LogService
+	// WaitHAKeeperState waits the specific hakeeper state.
 	WaitHAKeeperState(ctx context.Context, expected logpb.HAKeeperState)
-	// WaitClusterHealth(ctx context.Context)
-	// WaitShardByNum(typ string, batch int, ctx context.Context)
-	// WaitReplicaByNum(shardID uint64, batch int, ctx context.Context)
+
+	// WaitDNShardsReported waits the expected count of dn shards reported.
+	WaitDNShardsReported(ctx context.Context)
+	// WaitLogShardsReported waits the expected count of log shards reported.
+	WaitLogShardsReported(ctx context.Context)
+	// WaitDNReplicaReported waits dn replica reported.
+	WaitDNReplicaReported(ctx context.Context, shardID uint64)
+	// WaitLogReplicaReported waits log replicas reported.
+	WaitLogReplicaReported(ctx context.Context, shardID uint64)
 }
 
 // ----------------------------------------------------
@@ -233,63 +261,9 @@ func (c *testCluster) Close() error {
 	return nil
 }
 
-func (c *testCluster) GetDNService(uuid string) (DNService, error) {
-	c.dn.Lock()
-	defer c.dn.Unlock()
-
-	for i, cfg := range c.dn.cfgs {
-		if cfg.UUID == uuid {
-			return c.dn.svcs[i], nil
-		}
-	}
-	return nil, wrappedError(ErrServiceNotExist, uuid)
-}
-
-func (c *testCluster) GetLogService(uuid string) (LogService, error) {
-	c.log.Lock()
-	defer c.log.Unlock()
-
-	for _, svc := range c.log.svcs {
-		if svc.ID() == uuid {
-			return svc, nil
-		}
-	}
-	return nil, wrappedError(ErrServiceNotExist, uuid)
-}
-
-func (c *testCluster) GetDNServiceIndexed(index int) (DNService, error) {
-	c.dn.Lock()
-	defer c.dn.Unlock()
-
-	if index >= len(c.dn.svcs) || index < 0 {
-		return nil, wrappedError(
-			ErrInvalidServiceIndex, fmt.Sprintf("index: %d", index),
-		)
-	}
-	return c.dn.svcs[index], nil
-}
-
-func (c *testCluster) GetLogServiceIndexed(index int) (LogService, error) {
-	c.log.Lock()
-	defer c.log.Unlock()
-
-	if index >= len(c.log.svcs) || index < 0 {
-		return nil, wrappedError(
-			ErrInvalidServiceIndex, fmt.Sprintf("index: %d", index),
-		)
-	}
-	return c.log.svcs[index], nil
-}
-
-// NB: we could also fetch cluster state from non-leader hakeeper.
-func (c *testCluster) GetClusterState(
-	ctx context.Context,
-) (*logpb.CheckerState, error) {
-	c.WaitHAKeeperState(ctx, logpb.HAKeeperRunning)
-	leader := c.WaitHAKeeperLeader(ctx)
-	return leader.GetClusterState()
-}
-
+// ----------------------------------------------------------
+// The following are implements for interface `ClusterState`.
+// ----------------------------------------------------------
 func (c *testCluster) ListDNShards(
 	ctx context.Context,
 ) ([]metadata.DNShardRecord, error) {
@@ -324,6 +298,16 @@ func (c *testCluster) GetDNStoreInfo(
 	return logpb.DNStoreInfo{}, ErrServiceNotExist
 }
 
+func (c *testCluster) GetDNStoreInfoIndexed(
+	ctx context.Context, index int,
+) (logpb.DNStoreInfo, error) {
+	ds, err := c.GetDNServiceIndexed(index)
+	if err != nil {
+		return logpb.DNStoreInfo{}, err
+	}
+	return c.GetDNStoreInfo(ctx, ds.ID())
+}
+
 func (c *testCluster) GetLogStoreInfo(
 	ctx context.Context, uuid string,
 ) (logpb.LogStoreInfo, error) {
@@ -338,64 +322,95 @@ func (c *testCluster) GetLogStoreInfo(
 	return logpb.LogStoreInfo{}, ErrServiceNotExist
 }
 
-func (c *testCluster) GetDNStoreInfoIndexed(
+func (c *testCluster) GetLogStoreInfoIndexed(
 	ctx context.Context, index int,
-) (logpb.DNStoreInfo, error) {
-	state, err := c.GetClusterState(ctx)
+) (logpb.LogStoreInfo, error) {
+	ls, err := c.GetLogServiceIndexed(index)
 	if err != nil {
-		return logpb.DNStoreInfo{}, err
+		return logpb.LogStoreInfo{}, err
 	}
+	return c.GetLogStoreInfo(ctx, ls.ID())
+}
 
-	ds, err := c.GetDNServiceIndexed(index)
-	if err != nil {
-		return logpb.DNStoreInfo{}, err
-	}
-	uuid := ds.ID()
+func (c *testCluster) GetHAKeeperState() logpb.HAKeeperState {
+	state := c.getClusterState()
+	require.NotNil(c.t, state)
+	return state.State
+}
 
-	stores := state.DNState.Stores
-	if storeInfo, ok := stores[uuid]; ok {
-		return storeInfo, nil
+func (c *testCluster) GetHAKeeperConfig() hakeeper.Config {
+	return c.opt.BuildHAKeeperConfig()
+}
+
+func (c *testCluster) DNStoreExpired(uuid string) (bool, error) {
+	state := c.getClusterState()
+	require.NotNil(c.t, state)
+
+	dnStore, ok := state.DNState.Stores[uuid]
+	if !ok {
+		return false, wrappedError(ErrStoreNotReported, uuid)
 	}
-	return logpb.DNStoreInfo{}, ErrServiceNotExist
+
+	hkcfg := c.GetHAKeeperConfig()
+	return hkcfg.DnStoreExpired(dnStore.Tick, state.Tick), nil
 }
 
-func (c *testCluster) GetLogStoreInfoIndexed(
-	ctx context.Context, index int,
-) (logpb.LogStoreInfo, error) {
-	state, err := c.GetClusterState(ctx)
+func (c *testCluster) DNStoreExpiredIndexed(index int) (bool, error) {
+	ds, err := c.GetDNServiceIndexed(index)
 	if err != nil {
-		return logpb.LogStoreInfo{}, err
+		return false, err
 	}
+	return c.DNStoreExpired(ds.ID())
+}
 
-	ls, err := c.GetLogServiceIndexed(index)
-	if err != nil {
-		return logpb.LogStoreInfo{}, err
+func (c *testCluster) LogStoreExpired(uuid string) (bool, error) {
+	state := c.getClusterState()
+	require.NotNil(c.t, state)
+
+	logStore, ok := state.LogState.Stores[uuid]
+	if !ok {
+		return false, wrappedError(ErrStoreNotReported, uuid)
 	}
-	uuid := ls.ID()
 
-	stores := state.LogState.Stores
-	if storeInfo, ok := stores[uuid]; ok {
-		return storeInfo, nil
+	hkcfg := c.GetHAKeeperConfig()
+	return hkcfg.LogStoreExpired(logStore.Tick, state.Tick), nil
+}
+
+func (c *testCluster) LogStoreExpiredIndexed(index int) (bool, error) {
+	ls, err := c.GetLogServiceIndexed(index)
+	if err != nil {
+		return false, err
 	}
-	return logpb.LogStoreInfo{}, ErrServiceNotExist
+	return c.LogStoreExpired(ls.ID())
 }
 
-func (c *testCluster) GetHAKeeperState() logpb.HAKeeperState {
+func (c *testCluster) IsClusterHealthy() bool {
+	hkcfg := c.GetHAKeeperConfig()
 	state := c.getClusterState()
-	require.NotNil(c.t, state)
-	return state.State
+	_, healthy := syshealth.Check(
+		hkcfg,
+		state.GetClusterInfo(),
+		state.GetDNState(),
+		state.GetLogState(),
+		state.GetTick(),
+	)
+	return healthy
 }
 
+// --------------------------------------------------------------
+// The following are implements for interface `ClusterWaitState`.
+// --------------------------------------------------------------
 func (c *testCluster) WaitHAKeeperLeader(ctx context.Context) LogService {
 	for {
 		select {
 		case <-ctx.Done():
 			assert.FailNow(
 				c.t,
-				"terminated when waiting for hakeeper leader: %s", ctx.Err(),
+				"terminated when waiting for hakeeper leader",
+				"error: %s", ctx.Err(),
 			)
 		default:
-			time.Sleep(time.Millisecond * 100)
+			time.Sleep(defaultWaitInterval)
 
 			leader := c.getHAKeeperLeader()
 			if leader != nil {
@@ -404,6 +419,7 @@ func (c *testCluster) WaitHAKeeperLeader(ctx context.Context) LogService {
 		}
 	}
 }
+
 func (c *testCluster) WaitHAKeeperState(
 	ctx context.Context, expected logpb.HAKeeperState,
 ) {
@@ -412,10 +428,11 @@ func (c *testCluster) WaitHAKeeperState(
 		case <-ctx.Done():
 			assert.FailNow(
 				c.t,
-				"terminated when waiting for hakeeper state: %s", ctx.Err(),
+				"terminated when waiting for hakeeper state",
+				"error: %s", ctx.Err(),
 			)
 		default:
-			time.Sleep(100 * time.Millisecond)
+			time.Sleep(defaultWaitInterval)
 
 			state := c.getClusterState()
 			if state == nil {
@@ -428,6 +445,122 @@ func (c *testCluster) WaitHAKeeperState(
 	}
 }
 
+func (c *testCluster) WaitDNShardsReported(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			assert.FailNow(
+				c.t,
+				"terminated when waiting for all dn shards reported",
+				"error: %s", ctx.Err(),
+			)
+		default:
+			time.Sleep(defaultWaitInterval)
+
+			state := c.getClusterState()
+			if state == nil {
+				continue
+			}
+
+			expected := ParseExpectedDNShardCount(state.ClusterInfo)
+			reported := ParseReportedDNShardCount(
+				state.DNState, c.GetHAKeeperConfig(), state.Tick,
+			)
+			// FIXME: what about reported larger than expected
+			if reported >= expected {
+				return
+			}
+		}
+	}
+}
+
+func (c *testCluster) WaitLogShardsReported(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			assert.FailNow(
+				c.t,
+				"terminated when waiting for all log shards reported",
+				"error: %s", ctx.Err(),
+			)
+		default:
+			time.Sleep(defaultWaitInterval)
+
+			state := c.getClusterState()
+			if state == nil {
+				continue
+			}
+
+			expected := ParseExpectedLogShardCount(state.ClusterInfo)
+			reported := ParseReportedLogShardCount(
+				state.LogState, c.GetHAKeeperConfig(), state.Tick,
+			)
+			// FIXME: what about reported larger than expected
+			if reported >= expected {
+				return
+			}
+		}
+	}
+}
+
+func (c *testCluster) WaitDNReplicaReported(ctx context.Context, shardID uint64) {
+	for {
+		select {
+		case <-ctx.Done():
+			assert.FailNow(
+				c.t,
+				"terminated when waiting replica of dn shard reported",
+				"shard %d, error: %s", shardID, ctx.Err(),
+			)
+		default:
+			time.Sleep(defaultWaitInterval)
+
+			state := c.getClusterState()
+			if state == nil {
+				continue
+			}
+
+			reported := ParseDNShardReportedSize(
+				shardID, state.DNState, c.GetHAKeeperConfig(), state.Tick,
+			)
+			if reported >= DNShardExpectedSize {
+				return
+			}
+		}
+	}
+}
+
+func (c *testCluster) WaitLogReplicaReported(ctx context.Context, shardID uint64) {
+	for {
+		select {
+		case <-ctx.Done():
+			assert.FailNow(
+				c.t,
+				"terminated when waiting replica of log shard reported",
+				"shard %d, error: %s", shardID, ctx.Err(),
+			)
+		default:
+			time.Sleep(defaultWaitInterval)
+
+			state := c.getClusterState()
+			if state == nil {
+				continue
+			}
+
+			expected := ParseLogShardExpectedSize(shardID, state.ClusterInfo)
+			reported := ParseLogShardReportedSize(
+				shardID, state.LogState, c.GetHAKeeperConfig(), state.Tick,
+			)
+			if reported >= expected {
+				return
+			}
+		}
+	}
+}
+
+// --------------------------------------------------------------
+// The following are implements for interface `ClusterAwareness`.
+// --------------------------------------------------------------
 func (c *testCluster) ListDNServices() []string {
 	ids := make([]string, 0, len(c.dn.svcs))
 	for _, cfg := range c.dn.cfgs {
@@ -448,6 +581,66 @@ func (c *testCluster) ListHAKeeperServices() []LogService {
 	return c.selectHAkeeperServices()
 }
 
+func (c *testCluster) GetDNService(uuid string) (DNService, error) {
+	c.dn.Lock()
+	defer c.dn.Unlock()
+
+	for i, cfg := range c.dn.cfgs {
+		if cfg.UUID == uuid {
+			return c.dn.svcs[i], nil
+		}
+	}
+	return nil, wrappedError(ErrServiceNotExist, uuid)
+}
+
+func (c *testCluster) GetLogService(uuid string) (LogService, error) {
+	c.log.Lock()
+	defer c.log.Unlock()
+
+	for _, svc := range c.log.svcs {
+		if svc.ID() == uuid {
+			return svc, nil
+		}
+	}
+	return nil, wrappedError(ErrServiceNotExist, uuid)
+}
+
+func (c *testCluster) GetDNServiceIndexed(index int) (DNService, error) {
+	c.dn.Lock()
+	defer c.dn.Unlock()
+
+	if index >= len(c.dn.svcs) || index < 0 {
+		return nil, wrappedError(
+			ErrInvalidServiceIndex, fmt.Sprintf("index: %d", index),
+		)
+	}
+	return c.dn.svcs[index], nil
+}
+
+func (c *testCluster) GetLogServiceIndexed(index int) (LogService, error) {
+	c.log.Lock()
+	defer c.log.Unlock()
+
+	if index >= len(c.log.svcs) || index < 0 {
+		return nil, wrappedError(
+			ErrInvalidServiceIndex, fmt.Sprintf("index: %d", index),
+		)
+	}
+	return c.log.svcs[index], nil
+}
+
+// NB: we could also fetch cluster state from non-leader hakeeper.
+func (c *testCluster) GetClusterState(
+	ctx context.Context,
+) (*logpb.CheckerState, error) {
+	c.WaitHAKeeperState(ctx, logpb.HAKeeperRunning)
+	leader := c.WaitHAKeeperLeader(ctx)
+	return leader.GetClusterState()
+}
+
+// --------------------------------------------------------------
+// The following are implements for interface `ClusterOperation`.
+// --------------------------------------------------------------
 func (c *testCluster) CloseDNService(uuid string) error {
 	ds, err := c.GetDNService(uuid)
 	if err != nil {
@@ -694,6 +887,8 @@ func (c *testCluster) closeLogServices() error {
 }
 
 // getClusterState fetches cluster state from arbitrary hakeeper.
+//
+// NB: it's possible that getClusterState returns nil value.
 func (c *testCluster) getClusterState() *logpb.CheckerState {
 	var state *logpb.CheckerState
 	fn := func(index int, svc LogService) bool {
diff --git a/pkg/tests/service/service_test.go b/pkg/tests/service/service_test.go
index d618d6f67..6f2e798b0 100644
--- a/pkg/tests/service/service_test.go
+++ b/pkg/tests/service/service_test.go
@@ -24,7 +24,7 @@ import (
 	logpb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 )
 
-func TestCluster(t *testing.T) {
+func TestClusterStart(t *testing.T) {
 	// initialize cluster
 	c, err := NewCluster(t, DefaultOptions())
 	require.NoError(t, err)
@@ -36,17 +36,9 @@ func TestCluster(t *testing.T) {
 	// close the cluster
 	err = c.Close()
 	require.NoError(t, err)
-
-	// FIXME:
-	// 	- check cluster state via `ClusterAssertState`
-	// 	- wait cluster state via `ClusterWaitState`
 }
 
 func TestClusterAwareness(t *testing.T) {
-	// FIXME: skip this test after issue #4334 solved:
-	// https://github.com/matrixorigin/matrixone/issues/4334
-	t.Skip()
-
 	dnSvcNum := 2
 	logSvcNum := 3
 	opt := DefaultOptions().
@@ -76,6 +68,9 @@ func TestClusterAwareness(t *testing.T) {
 	lsuuids := c.ListLogServices()
 	require.Equal(t, logSvcNum, len(lsuuids))
 
+	hksvcs := c.ListHAKeeperServices()
+	require.NotZero(t, len(hksvcs))
+
 	dn, err := c.GetDNService(dsuuids[0])
 	require.NoError(t, err)
 	require.Equal(t, ServiceStarted, dn.Status())
@@ -89,8 +84,7 @@ func TestClusterAwareness(t *testing.T) {
 	leader := c.WaitHAKeeperLeader(ctx1)
 	require.NotNil(t, leader)
 
-	// we must wait for hakeeper's running state,
-	// or hakeeper wouldn't receive hearbeat.
+	// we must wait for hakeeper's running state, or hakeeper wouldn't receive hearbeat.
 	ctx2, cancel2 := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel2()
 	c.WaitHAKeeperState(ctx2, logpb.HAKeeperRunning)
@@ -127,10 +121,11 @@ func TestClusterOperation(t *testing.T) {
 	// -------------------------------------------
 	// the following would test `ClusterOperation`
 	// -------------------------------------------
-	// 1. close dn services by different ways
+
+	// 1. start/close dn services via different ways
 	dsuuids := c.ListDNServices()
 	require.Equal(t, dnSvcNum, len(dsuuids))
-	// 1.a close dn service by uuid
+	// 1.a start/close dn service by uuid
 	{
 		index := 0
 		dsuuid := dsuuids[index]
@@ -140,13 +135,18 @@ func TestClusterOperation(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, ServiceStarted, ds.Status())
 
+		// start it
+		err = c.StartDNService(dsuuid)
+		require.NoError(t, err)
+		require.Equal(t, ServiceStarted, ds.Status())
+
 		// close it
 		err = c.CloseDNService(dsuuid)
 		require.NoError(t, err)
 		require.Equal(t, ServiceClosed, ds.Status())
 	}
 
-	// 1.b close dn service by index
+	// 1.b start/close dn service by index
 	{
 		index := 1
 
@@ -155,13 +155,18 @@ func TestClusterOperation(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, ServiceStarted, ds.Status())
 
+		// start it
+		err = c.StartDNServiceIndexed(index)
+		require.NoError(t, err)
+		require.Equal(t, ServiceStarted, ds.Status())
+
 		// close it
 		err = c.CloseDNServiceIndexed(index)
 		require.NoError(t, err)
 		require.Equal(t, ServiceClosed, ds.Status())
 	}
 
-	// 1.c close dn service by instance
+	// 1.c start/close dn service by instance
 	{
 		index := 2
 
@@ -170,16 +175,21 @@ func TestClusterOperation(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, ServiceStarted, ds.Status())
 
+		// start it
+		err = ds.Start()
+		require.NoError(t, err)
+		require.Equal(t, ServiceStarted, ds.Status())
+
 		// close it
 		err = ds.Close()
 		require.NoError(t, err)
 		require.Equal(t, ServiceClosed, ds.Status())
 	}
 
-	// 2. close log services by different ways
+	// 2. start/close log services by different ways
 	lsuuids := c.ListLogServices()
 	require.Equal(t, logSvcNum, len(lsuuids))
-	// 2.a close log service by uuid
+	// 2.a start/close log service by uuid
 	{
 		index := 0
 		lsuuid := lsuuids[index]
@@ -189,13 +199,18 @@ func TestClusterOperation(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, ServiceStarted, ls.Status())
 
+		// start it
+		err = c.StartLogService(lsuuid)
+		require.NoError(t, err)
+		require.Equal(t, ServiceStarted, ls.Status())
+
 		// close it
 		err = c.CloseLogService(lsuuid)
 		require.NoError(t, err)
 		require.Equal(t, ServiceClosed, ls.Status())
 	}
 
-	// 2.b close log service by index
+	// 2.b start/close log service by index
 	{
 		index := 1
 
@@ -204,13 +219,18 @@ func TestClusterOperation(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, ServiceStarted, ls.Status())
 
+		// start it
+		err = c.StartLogServiceIndexed(index)
+		require.NoError(t, err)
+		require.Equal(t, ServiceStarted, ls.Status())
+
 		// close it
 		err = c.CloseLogServiceIndexed(index)
 		require.NoError(t, err)
 		require.Equal(t, ServiceClosed, ls.Status())
 	}
 
-	// 2.c close log service by instance
+	// 2.c start/close log service by instance
 	{
 		index := 2
 
@@ -219,6 +239,11 @@ func TestClusterOperation(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, ServiceStarted, ls.Status())
 
+		// start it
+		err = ls.Start()
+		require.NoError(t, err)
+		require.Equal(t, ServiceStarted, ls.Status())
+
 		// close it
 		err = ls.Close()
 		require.NoError(t, err)
@@ -227,10 +252,6 @@ func TestClusterOperation(t *testing.T) {
 }
 
 func TestClusterState(t *testing.T) {
-	// FIXME: skip this test after issue #4334 solved:
-	// https://github.com/matrixorigin/matrixone/issues/4334
-	t.Skip()
-
 	dnSvcNum := 2
 	logSvcNum := 3
 	opt := DefaultOptions().
@@ -251,9 +272,9 @@ func TestClusterState(t *testing.T) {
 		require.NoError(t, err)
 	}()
 
-	// -------------------------------------------
-	// the following would test `ClusterState`
-	// -------------------------------------------
+	// ----------------------------------------
+	// the following would test `ClusterState`.
+	// ----------------------------------------
 	ctx1, cancel1 := context.WithTimeout(context.Background(), 2*time.Second)
 	defer cancel1()
 	leader := c.WaitHAKeeperLeader(ctx1)
@@ -265,12 +286,17 @@ func TestClusterState(t *testing.T) {
 	lsuuids := c.ListLogServices()
 	require.Equal(t, logSvcNum, len(lsuuids))
 
-	// we must wait for hakeeper's running state,
-	// or hakeeper wouldn't receive hearbeat.
-	ctx2, cancel2 := context.WithTimeout(context.Background(), 2*time.Second)
+	// we must wait for hakeeper's running state, or hakeeper wouldn't receive hearbeat.
+	ctx2, cancel2 := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel2()
 	c.WaitHAKeeperState(ctx2, logpb.HAKeeperRunning)
 
+	hkstate := c.GetHAKeeperState()
+	require.Equal(t, logpb.HAKeeperRunning, hkstate)
+
+	// cluster should be healthy
+	require.True(t, c.IsClusterHealthy())
+
 	ctx3, cancel3 := context.WithTimeout(context.Background(), 2*time.Second)
 	defer cancel3()
 	state, err := c.GetClusterState(ctx3)
@@ -290,32 +316,134 @@ func TestClusterState(t *testing.T) {
 	_, err = c.ListLogShards(ctx5)
 	require.NoError(t, err)
 
-	// test GetDNStoreInfo and GetDNStoreInfoIndexed
-	dnIndex := 0
-	dsuuid := dsuuids[dnIndex]
+	// test for:
+	//   - GetDNStoreInfo
+	//   - GetDNStoreInfoIndexed
+	//   - DNStoreExpired
+	//   - DNStoreExpiredIndexed
+	{
+		dnIndex := 0
+		dsuuid := dsuuids[dnIndex]
 
-	ctx6, cancel6 := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel6()
-	dnStoreInfo1, err := c.GetDNStoreInfo(ctx6, dsuuid)
-	require.NoError(t, err)
+		ctx6, cancel6 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel6()
+		dnStoreInfo1, err := c.GetDNStoreInfo(ctx6, dsuuid)
+		require.NoError(t, err)
 
-	ctx7, cancel7 := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel7()
-	dnStoreInfo2, err := c.GetDNStoreInfoIndexed(ctx7, dnIndex)
-	require.NoError(t, err)
-	require.Equal(t, dnStoreInfo1, dnStoreInfo2)
+		ctx7, cancel7 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel7()
+		dnStoreInfo2, err := c.GetDNStoreInfoIndexed(ctx7, dnIndex)
+		require.NoError(t, err)
+		require.Equal(t, dnStoreInfo1.Shards, dnStoreInfo2.Shards)
+
+		expired1, err := c.DNStoreExpired(dsuuid)
+		require.NoError(t, err)
+		require.False(t, expired1)
+
+		expired2, err := c.DNStoreExpiredIndexed(dnIndex)
+		require.NoError(t, err)
+		require.False(t, expired2)
+	}
+
+	// test for:
+	//   - GetLogStoreInfo
+	//   - GetLogStoreInfoIndexed
+	//   - LogStoreExpired
+	//   - LogStoreExpiredIndexed
+	{
+		logIndex := 1
+		lsuuid := lsuuids[logIndex]
+
+		ctx8, cancel8 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel8()
+		logStoreInfo1, err := c.GetLogStoreInfo(ctx8, lsuuid)
+		require.NoError(t, err)
 
-	logIndex := 1
-	lsuuid := lsuuids[logIndex]
+		ctx9, cancel9 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel9()
+		logStoreInfo2, err := c.GetLogStoreInfoIndexed(ctx9, logIndex)
+		require.NoError(t, err)
+		require.Equal(t, logStoreInfo1.Replicas, logStoreInfo2.Replicas)
+
+		expired1, err := c.LogStoreExpired(lsuuid)
+		require.NoError(t, err)
+		require.False(t, expired1)
 
-	ctx8, cancel8 := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel8()
-	logStoreInfo1, err := c.GetLogStoreInfo(ctx8, lsuuid)
+		expired2, err := c.LogStoreExpiredIndexed(logIndex)
+		require.NoError(t, err)
+		require.False(t, expired2)
+	}
+}
+
+func TestClusterWaitState(t *testing.T) {
+	dnSvcNum := 2
+	logSvcNum := 3
+	opt := DefaultOptions().
+		WithDNServiceNum(dnSvcNum).
+		WithLogServiceNum(logSvcNum)
+
+	// initialize cluster
+	c, err := NewCluster(t, opt)
 	require.NoError(t, err)
 
-	ctx9, cancel9 := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel9()
-	logStoreInfo2, err := c.GetLogStoreInfoIndexed(ctx9, logIndex)
+	// start the cluster
+	err = c.Start()
 	require.NoError(t, err)
-	require.Equal(t, logStoreInfo1, logStoreInfo2)
+
+	// close the cluster after all
+	defer func() {
+		err := c.Close()
+		require.NoError(t, err)
+	}()
+
+	// --------------------------------------------
+	// the following would test `ClusterWaitState`.
+	// --------------------------------------------
+
+	// we must wait for hakeeper's running state, or hakeeper wouldn't receive hearbeat.
+	ctx1, cancel1 := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel1()
+	c.WaitHAKeeperState(ctx1, logpb.HAKeeperRunning)
+
+	// test WaitDNShardsReported
+	{
+		ctx2, cancel2 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel2()
+		c.WaitDNShardsReported(ctx2)
+	}
+
+	// test WaitLogShardsReported
+	{
+		ctx3, cancel3 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel3()
+		c.WaitLogShardsReported(ctx3)
+	}
+
+	// test WaitDNReplicaReported
+	{
+		ctx4, cancel4 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel4()
+		dnShards, err := c.ListDNShards(ctx4)
+		require.NoError(t, err)
+		require.NotZero(t, len(dnShards))
+
+		dnShardID := dnShards[0].ShardID
+		ctx5, cancel5 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel5()
+		c.WaitDNReplicaReported(ctx5, dnShardID)
+	}
+
+	// test WaitLogReplicaReported
+	{
+		ctx6, cancel6 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel6()
+		logShards, err := c.ListLogShards(ctx6)
+		require.NotZero(t, len(logShards))
+		require.NoError(t, err)
+
+		logShardID := logShards[0].ShardID
+		ctx7, cancel7 := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel7()
+		c.WaitLogReplicaReported(ctx7, logShardID)
+	}
 }
diff --git a/pkg/tests/service/utility.go b/pkg/tests/service/utility.go
new file mode 100644
index 000000000..efc49d68f
--- /dev/null
+++ b/pkg/tests/service/utility.go
@@ -0,0 +1,132 @@
+// 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 service
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+)
+
+const (
+	// The expected number of dn replicas.
+	DNShardExpectedSize = 1
+)
+
+// ParseExpectedDNShardCount returns the expected count of dn shards.
+func ParseExpectedDNShardCount(cluster pb.ClusterInfo) int {
+	set := make(map[uint64]struct{})
+	for _, record := range cluster.DNShards {
+		set[record.ShardID] = struct{}{}
+
+	}
+	return len(set)
+}
+
+// ParseExpectedLogShardCount returns the expected count of log shards.
+func ParseExpectedLogShardCount(cluster pb.ClusterInfo) int {
+	set := make(map[uint64]struct{})
+	for _, record := range cluster.LogShards {
+		set[record.ShardID] = struct{}{}
+
+	}
+	return len(set)
+}
+
+// ParseReportedDNShardCount returns the reported count of dn shards.
+func ParseReportedDNShardCount(
+	state pb.DNState, hkcfg hakeeper.Config, currTick uint64,
+) int {
+	set := make(map[uint64]struct{})
+	for _, storeInfo := range state.Stores {
+		// ignore expired dn stores
+		if hkcfg.DnStoreExpired(storeInfo.Tick, currTick) {
+			continue
+		}
+
+		// record dn shard
+		for _, shardInfo := range storeInfo.Shards {
+			set[shardInfo.ShardID] = struct{}{}
+		}
+	}
+	return len(set)
+}
+
+// ParseReportedLogShardCount returns the reported count of log shards.
+func ParseReportedLogShardCount(
+	state pb.LogState, hkcfg hakeeper.Config, currTick uint64,
+) int {
+	set := make(map[uint64]struct{})
+	for _, storeInfo := range state.Stores {
+		// ignore expired log stores
+		if hkcfg.LogStoreExpired(storeInfo.Tick, currTick) {
+			continue
+		}
+
+		for _, replicaInfo := range storeInfo.Replicas {
+			set[replicaInfo.ShardID] = struct{}{}
+		}
+	}
+	return len(set)
+}
+
+// ParseLogShardExpectedSize returns the expected count of log replicas.
+func ParseLogShardExpectedSize(shardID uint64, cluster pb.ClusterInfo) int {
+	for _, record := range cluster.LogShards {
+		if record.ShardID == shardID {
+			return int(record.NumberOfReplicas)
+		}
+	}
+	return 0
+}
+
+// ParseLogShardReportedSize returns the reported count of log replicas.
+func ParseLogShardReportedSize(
+	shardID uint64, state pb.LogState, hkcfg hakeeper.Config, currTick uint64,
+) int {
+	set := make(map[uint64]struct{})
+	for _, storeInfo := range state.Stores {
+		// ignore expired log stores
+		if hkcfg.LogStoreExpired(storeInfo.Tick, currTick) {
+			continue
+		}
+
+		for _, replicaInfo := range storeInfo.Replicas {
+			if replicaInfo.ShardID == shardID {
+				set[replicaInfo.ReplicaID] = struct{}{}
+			}
+		}
+	}
+	return len(set)
+}
+
+// ParseDNShardReportedSize returns the reported count of dn replicas.
+func ParseDNShardReportedSize(
+	shardID uint64, state pb.DNState, hkcfg hakeeper.Config, currTick uint64,
+) int {
+	set := make(map[uint64]struct{})
+	for _, storeInfo := range state.Stores {
+		// ignore expired dn stores
+		if hkcfg.DnStoreExpired(storeInfo.Tick, currTick) {
+			continue
+		}
+
+		for _, shardInfo := range storeInfo.Shards {
+			if shardInfo.ShardID == shardID {
+				set[shardInfo.ReplicaID] = struct{}{}
+			}
+		}
+	}
+	return len(set)
+}
diff --git a/pkg/tests/service/utility_test.go b/pkg/tests/service/utility_test.go
new file mode 100644
index 000000000..d92522cac
--- /dev/null
+++ b/pkg/tests/service/utility_test.go
@@ -0,0 +1,235 @@
+// 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 service
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/matrixorigin/matrixone/pkg/hakeeper"
+	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
+	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
+)
+
+const (
+	// default replica number for log shard
+	defaultLogShardSize = 3
+)
+
+func TestParseExpectedDNShardCount(t *testing.T) {
+	cluster := pb.ClusterInfo{
+		DNShards: mockDNShardRecords(10, 11, 12, 12),
+	}
+	// expected dn shards: 10, 11, 12
+	require.Equal(t, 3, ParseExpectedDNShardCount(cluster))
+}
+
+func TestParseExpectedLogShardCount(t *testing.T) {
+	cluster := pb.ClusterInfo{
+		LogShards: mockLogShardRecords(10, 11, 11),
+	}
+	// expected log shards: 10, 11
+	require.Equal(t, 2, ParseExpectedLogShardCount(cluster))
+}
+
+func TestParseReportedDNShardCount(t *testing.T) {
+	hkcfg := hakeeper.Config{}
+	hkcfg.Fill()
+
+	expiredTick := uint64(10)
+	currTick := hkcfg.ExpiredTick(expiredTick, hkcfg.DNStoreTimeout) + 1
+
+	dnState := pb.DNState{
+		Stores: map[string]pb.DNStoreInfo{
+			"expired1": {
+				Tick: expiredTick,
+				Shards: []pb.DNShardInfo{
+					mockDnShardInfo(10, 100),
+				},
+			},
+			"working1": {
+				Tick: currTick,
+				Shards: []pb.DNShardInfo{
+					mockDnShardInfo(11, 101),
+					mockDnShardInfo(11, 102),
+					mockDnShardInfo(12, 103),
+				},
+			},
+		},
+	}
+
+	// working dn shards: 11, 12
+	require.Equal(t, 2, ParseReportedDNShardCount(dnState, hkcfg, currTick))
+}
+
+func TestParseReportedLogShardCount(t *testing.T) {
+	hkcfg := hakeeper.Config{}
+	hkcfg.Fill()
+
+	expiredTick := uint64(10)
+	currTick := hkcfg.ExpiredTick(expiredTick, hkcfg.LogStoreTimeout) + 1
+
+	logState := pb.LogState{
+		Stores: map[string]pb.LogStoreInfo{
+			"expired1": {
+				Tick: expiredTick,
+				Replicas: []pb.LogReplicaInfo{
+					mockLogReplicaInfo(11, 100),
+				},
+			},
+			"workding1": {
+				Tick: currTick,
+				Replicas: []pb.LogReplicaInfo{
+					mockLogReplicaInfo(11, 101),
+				},
+			},
+			"workding2": {
+				Tick: currTick,
+				Replicas: []pb.LogReplicaInfo{
+					mockLogReplicaInfo(12, 102),
+				},
+			},
+			"workding3": {
+				Tick: currTick,
+				Replicas: []pb.LogReplicaInfo{
+					mockLogReplicaInfo(12, 103),
+				},
+			},
+		},
+	}
+
+	// reported working log shard: 11, 12
+	require.Equal(t, 2, ParseReportedLogShardCount(logState, hkcfg, currTick))
+}
+
+func TestParseLogShardExpectedSize(t *testing.T) {
+	cluster := pb.ClusterInfo{
+		LogShards: mockLogShardRecords(10, 11),
+	}
+
+	// log shard 10 recorded in ClusterInfo
+	require.Equal(t, defaultLogShardSize, ParseLogShardExpectedSize(10, cluster))
+
+	// log shard 100 not recorded in ClusterInfo
+	require.Equal(t, 0, ParseLogShardExpectedSize(100, cluster))
+}
+
+func TestParseLogShardReportedSize(t *testing.T) {
+	hkcfg := hakeeper.Config{}
+	hkcfg.Fill()
+
+	expiredTick := uint64(10)
+	currTick := hkcfg.ExpiredTick(expiredTick, hkcfg.LogStoreTimeout) + 1
+
+	logState := pb.LogState{
+		Stores: map[string]pb.LogStoreInfo{
+			"expired1": {
+				Tick: expiredTick,
+				Replicas: []pb.LogReplicaInfo{
+					mockLogReplicaInfo(11, 100),
+				},
+			},
+			"workding1": {
+				Tick: currTick,
+				Replicas: []pb.LogReplicaInfo{
+					mockLogReplicaInfo(11, 101),
+				},
+			},
+			"workding2": {
+				Tick: currTick,
+				Replicas: []pb.LogReplicaInfo{
+					mockLogReplicaInfo(11, 102),
+				},
+			},
+			"workding3": {
+				Tick: currTick,
+				Replicas: []pb.LogReplicaInfo{
+					mockLogReplicaInfo(11, 103),
+				},
+			},
+		},
+	}
+
+	// 3 working replicas reported for log shard 11
+	require.Equal(t, 3, ParseLogShardReportedSize(11, logState, hkcfg, currTick))
+}
+
+func TestParseDNShardReportedSize(t *testing.T) {
+	hkcfg := hakeeper.Config{}
+	hkcfg.Fill()
+
+	expiredTick := uint64(10)
+	currTick := hkcfg.ExpiredTick(expiredTick, hkcfg.DNStoreTimeout) + 1
+
+	dnState := pb.DNState{
+		Stores: map[string]pb.DNStoreInfo{
+			"expired1": {
+				Tick: expiredTick,
+				Shards: []pb.DNShardInfo{
+					mockDnShardInfo(10, 100),
+				},
+			},
+			"working1": {
+				Tick: currTick,
+				Shards: []pb.DNShardInfo{
+					mockDnShardInfo(11, 101),
+					mockDnShardInfo(11, 102),
+					mockDnShardInfo(12, 103),
+				},
+			},
+		},
+	}
+
+	require.Equal(t, 2, ParseDNShardReportedSize(11, dnState, hkcfg, currTick))
+}
+
+func mockDNShardRecords(ids ...uint64) []metadata.DNShardRecord {
+	records := make([]metadata.DNShardRecord, 0, len(ids))
+	for _, id := range ids {
+		records = append(records, metadata.DNShardRecord{
+			ShardID:    id,
+			LogShardID: id,
+		})
+	}
+	return records
+}
+
+func mockLogShardRecords(ids ...uint64) []metadata.LogShardRecord {
+	records := make([]metadata.LogShardRecord, 0, len(ids))
+	for _, id := range ids {
+		records = append(records, metadata.LogShardRecord{
+			ShardID:          id,
+			NumberOfReplicas: uint64(defaultLogShardSize),
+		})
+	}
+	return records
+}
+
+func mockDnShardInfo(shardID, replicaID uint64) pb.DNShardInfo {
+	return pb.DNShardInfo{
+		ShardID:   shardID,
+		ReplicaID: replicaID,
+	}
+}
+
+func mockLogReplicaInfo(shardID, replicaID uint64) pb.LogReplicaInfo {
+	return pb.LogReplicaInfo{
+		ReplicaID: replicaID,
+		LogShardInfo: pb.LogShardInfo{
+			ShardID: shardID,
+		},
+	}
+}