Merge remote-tracking branch 'origin/incr-validator' into incr-validator

pingcap · Mar 3, 2022 · 9533c4d · 9533c4d
2 parents 6277f06 + 9df2d93
commit 9533c4d
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 12 deletions.
diff --git a/cdc/server.go b/cdc/server.go
@@ -257,9 +257,8 @@ func (s *Server) etcdHealthChecker(ctx context.Context) error {
 			for _, pdEndpoint := range s.pdEndpoints {
 				start := time.Now()
 				ctx, cancel := context.WithTimeout(ctx, time.Second*10)
-				// TODO: PD has removed "/health" API.
 				req, err := http.NewRequestWithContext(
-					ctx, http.MethodGet, fmt.Sprintf("%s/health", pdEndpoint), nil)
+					ctx, http.MethodGet, fmt.Sprintf("%s/pd/api/v1/health", pdEndpoint), nil)
 				if err != nil {
 					log.Warn("etcd health check failed", zap.Error(err))
 					cancel()

diff --git a/cdc/sink/mysql_test.go b/cdc/sink/mysql_test.go
@@ -23,7 +23,6 @@ import (
 	"sort"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/DATA-DOG/go-sqlmock"
 	dmysql "github.com/go-sql-driver/mysql"
@@ -1128,7 +1127,7 @@ func TestMySQLSinkClose(t *testing.T) {
 	require.Nil(t, err)
 }
 
-func TestMySQLSinkFlushResovledTs(t *testing.T) {
+func TestMySQLSinkFlushResolvedTs(t *testing.T) {
 	dbIndex := 0
 	mockGetDBConn := func(ctx context.Context, dsnStr string) (*sql.DB, error) {
 		defer func() {
@@ -1177,7 +1176,7 @@ func TestMySQLSinkFlushResovledTs(t *testing.T) {
 	require.Nil(t, err)
 	checkpoint, err := sink.FlushRowChangedEvents(ctx, model.TableID(1), 1)
 	require.Nil(t, err)
-	require.Equal(t, uint64(0), checkpoint)
+	require.True(t, checkpoint <= 1)
 	rows := []*model.RowChangedEvent{
 		{
 			Table:    &model.TableName{Schema: "s1", Table: "t1", TableID: 1},
@@ -1190,10 +1189,9 @@ func TestMySQLSinkFlushResovledTs(t *testing.T) {
 	err = sink.EmitRowChangedEvents(ctx, rows...)
 	require.Nil(t, err)
 	checkpoint, err = sink.FlushRowChangedEvents(ctx, model.TableID(1), 6)
-	require.True(t, checkpoint <= 5)
-	time.Sleep(500 * time.Millisecond)
+	require.True(t, checkpoint <= 6)
 	require.Nil(t, err)
-	require.Equal(t, uint64(6), sink.getTableCheckpointTs(model.TableID(1)))
+	require.True(t, sink.getTableCheckpointTs(model.TableID(1)) <= 6)
 	rows = []*model.RowChangedEvent{
 		{
 			Table:    &model.TableName{Schema: "s1", Table: "t2", TableID: 2},
@@ -1207,9 +1205,7 @@ func TestMySQLSinkFlushResovledTs(t *testing.T) {
 	require.Nil(t, err)
 	checkpoint, err = sink.FlushRowChangedEvents(ctx, model.TableID(2), 5)
 	require.True(t, checkpoint <= 5)
-	time.Sleep(500 * time.Millisecond)
-	require.Nil(t, err)
-	require.Equal(t, uint64(5), sink.getTableCheckpointTs(model.TableID(2)))
-	err = sink.Close(ctx)
 	require.Nil(t, err)
+	require.True(t, sink.getTableCheckpointTs(model.TableID(2)) <= 5)
+	_ = sink.Close(ctx)
 }