ddl: refine ddl log and add more useful message in log (#3223)

ref #3178

Author: hongyunyan

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -554,7 +554,7 @@ func (d *BasicDispatcher) HandleDispatcherStatus(dispatcherStatus *heartbeatpb.D
 		pendingEvent := d.blockEventStatus.getEvent()
 		if pendingEvent == nil && action.CommitTs > d.GetResolvedTs() {
 			// we have not received the block event, and the action is for the future event, so just ignore
-			log.Info("pending event is nil, and the action's commit is larger than dispatchers resolvedTs",
+			log.Debug("pending event is nil, and the action's commit is larger than dispatchers resolvedTs",
 				zap.Uint64("resolvedTs", d.GetResolvedTs()),
 				zap.Uint64("actionCommitTs", action.CommitTs),
 				zap.Stringer("dispatcher", d.id))
@@ -592,7 +592,7 @@ func (d *BasicDispatcher) HandleDispatcherStatus(dispatcherStatus *heartbeatpb.D
 		} else {
 			ts, ok := d.blockEventStatus.getEventCommitTs()
 			if ok && action.CommitTs > ts {
-				log.Info("pending event's commitTs is smaller than the action's commitTs, just ignore it",
+				log.Debug("pending event's commitTs is smaller than the action's commitTs, just ignore it",
 					zap.Uint64("pendingEventCommitTs", ts),
 					zap.Uint64("actionCommitTs", action.CommitTs),
 					zap.Stringer("dispatcher", d.id))

downstreamadapter/dispatcher/helper.go

@@ -261,21 +261,23 @@ type ResendTask struct {
 	taskHandle *threadpool.TaskHandle
 }
 
+const resendTimeInterval = 10 * time.Second
+
 func newResendTask(message *heartbeatpb.TableSpanBlockStatus, dispatcher Dispatcher, callback func()) *ResendTask {
 	taskScheduler := GetDispatcherTaskScheduler()
 	t := &ResendTask{
 		message:    message,
 		dispatcher: dispatcher,
 		callback:   callback,
 	}
-	t.taskHandle = taskScheduler.Submit(t, time.Now().Add(50*time.Millisecond))
+	t.taskHandle = taskScheduler.Submit(t, time.Now().Add(resendTimeInterval))
 	return t
 }
 
 func (t *ResendTask) Execute() time.Time {
 	log.Debug("resend task", zap.Any("message", t.message), zap.Any("dispatcherID", t.dispatcher.GetId()))
 	t.dispatcher.GetBlockStatusesChan() <- t.message
-	return time.Now().Add(10 * time.Second)
+	return time.Now().Add(resendTimeInterval)
 }
 
 func (t *ResendTask) Cancel() {

maintainer/barrier.go

@@ -347,7 +347,7 @@ func (b *Barrier) handleEventDone(changefeedID common.ChangeFeedID, dispatcherID
 	key := getEventKey(status.State.BlockTs, status.State.IsSyncPoint)
 	event, ok := b.blockedEvents.Get(key)
 	if !ok {
-		log.Info("No block event found, ignore the event done message",
+		log.Debug("No block event found, ignore the event done message",
 			zap.String("changefeed", changefeedID.Name()),
 			zap.String("dispatcher", dispatcherID.String()),
 			zap.Uint64("commitTs", status.State.BlockTs),
@@ -392,7 +392,7 @@ func (b *Barrier) handleBlockState(changefeedID common.ChangeFeedID,
 		}
 		if event.selected.Load() {
 			// the event already in the selected state, ignore the block event just sent ack
-			log.Warn("the block event already selected, ignore the block event",
+			log.Debug("the block event already selected, ignore the block event",
 				zap.String("changefeed", changefeedID.Name()),
 				zap.String("dispatcher", dispatcherID.String()),
 				zap.Uint64("commitTs", blockState.BlockTs),

maintainer/barrier_event.go

@@ -106,7 +106,7 @@ func NewBlockEvent(cfID common.ChangeFeedID,
 			if dynamicSplitEnabled {
 				event.rangeChecker = range_checker.NewTableSpanRangeChecker(spanController.GetkeyspaceID(), status.BlockTables.TableIDs)
 			} else {
-				event.rangeChecker = range_checker.NewTableCountChecker(len(status.BlockTables.TableIDs))
+				event.rangeChecker = range_checker.NewTableCountChecker(status.BlockTables.TableIDs)
 			}
 		}
 	}
@@ -120,33 +120,31 @@ func NewBlockEvent(cfID common.ChangeFeedID,
 }
 
 func (be *BarrierEvent) createRangeCheckerForTypeAll() {
+	reps := be.spanController.GetAllTasks()
+	tbls := make([]int64, 0, len(reps))
+	for _, rep := range reps {
+		tbls = append(tbls, rep.Span.TableID)
+	}
 	if be.dynamicSplitEnabled {
-		reps := be.spanController.GetAllTasks()
-		tbls := make([]int64, 0, len(reps))
-		for _, rep := range reps {
-			tbls = append(tbls, rep.Span.TableID)
-		}
-		tbls = append(tbls, common.DDLSpanTableID)
 		be.rangeChecker = range_checker.NewTableSpanRangeChecker(be.spanController.GetkeyspaceID(), tbls)
 	} else {
-		be.rangeChecker = range_checker.NewTableCountChecker(be.spanController.TaskSize())
+		be.rangeChecker = range_checker.NewTableCountChecker(tbls)
 	}
 	log.Info("create range checker for block event", zap.Any("influcenceType", be.blockedDispatchers.InfluenceType), zap.Any("commitTs", be.commitTs))
 }
 
 func (be *BarrierEvent) createRangeCheckerForTypeDB() {
-	if be.dynamicSplitEnabled {
-		reps := be.spanController.GetTasksBySchemaID(be.blockedDispatchers.SchemaID)
-		tbls := make([]int64, 0, len(reps))
-		for _, rep := range reps {
-			tbls = append(tbls, rep.Span.TableID)
-		}
+	reps := be.spanController.GetTasksBySchemaID(be.blockedDispatchers.SchemaID)
+	tbls := make([]int64, 0, len(reps))
+	for _, rep := range reps {
+		tbls = append(tbls, rep.Span.TableID)
+	}
 
-		tbls = append(tbls, common.DDLSpanTableID)
+	tbls = append(tbls, common.DDLSpanTableID)
+	if be.dynamicSplitEnabled {
 		be.rangeChecker = range_checker.NewTableSpanRangeChecker(be.spanController.GetkeyspaceID(), tbls)
 	} else {
-		be.rangeChecker = range_checker.NewTableCountChecker(
-			be.spanController.GetTaskSizeBySchemaID(be.blockedDispatchers.SchemaID) + 1 /*table trigger event dispatcher*/)
+		be.rangeChecker = range_checker.NewTableCountChecker(tbls)
 	}
 	log.Info("create range checker for block event", zap.Any("influcenceType", be.blockedDispatchers.InfluenceType), zap.Any("commitTs", be.commitTs))
 }
@@ -363,7 +361,7 @@ func (be *BarrierEvent) allDispatcherReported() bool {
 				if be.dynamicSplitEnabled {
 					be.rangeChecker = range_checker.NewTableSpanRangeChecker(be.spanController.GetkeyspaceID(), be.blockedDispatchers.TableIDs)
 				} else {
-					be.rangeChecker = range_checker.NewTableCountChecker(len(be.blockedDispatchers.TableIDs))
+					be.rangeChecker = range_checker.NewTableCountChecker(be.blockedDispatchers.TableIDs)
 				}
 			case heartbeatpb.InfluenceType_DB:
 				be.createRangeCheckerForTypeDB()
@@ -515,7 +513,7 @@ func (be *BarrierEvent) checkBlockedDispatchers() {
 }
 
 func (be *BarrierEvent) resend(mode int64) []*messaging.TargetMessage {
-	if time.Since(be.lastResendTime) < 10*time.Second {
+	if time.Since(be.lastResendTime) < time.Second {
 		return nil
 	}
 	var msgs []*messaging.TargetMessage

maintainer/barrier_event_test.go

@@ -13,7 +13,23 @@
 
 package maintainer
 
-/*
+import (
+	"testing"
+	"time"
+
+	"github.com/pingcap/ticdc/heartbeatpb"
+	"github.com/pingcap/ticdc/maintainer/operator"
+	"github.com/pingcap/ticdc/maintainer/replica"
+	"github.com/pingcap/ticdc/maintainer/span"
+	"github.com/pingcap/ticdc/maintainer/testutil"
+	"github.com/pingcap/ticdc/pkg/common"
+	appcontext "github.com/pingcap/ticdc/pkg/common/context"
+	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
+	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/server/watcher"
+	"github.com/stretchr/testify/require"
+)
+
 func TestScheduleEvent(t *testing.T) {
 	testutil.SetUpTestServices()
 	tableTriggerEventDispatcherID := common.NewDispatcherID()
@@ -209,4 +225,3 @@ func TestUpdateSchemaID(t *testing.T) {
 	require.Len(t, spanController.GetTasksBySchemaID(2), 1)
 	require.Equal(t, spanController.GetTasksByTableID(1)[0].GetSchemaID(), int64(2))
 }
-*/