fix integration test.

cdc/model/schema_storage.go

@@ -16,11 +16,13 @@ package model
 import (
 	"fmt"
 
+	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/parser/types"
 	"github.com/pingcap/tidb/table/tables"
 	"github.com/pingcap/tidb/util/rowcodec"
+	"go.uber.org/zap"
 )
 
 const (
@@ -318,6 +320,9 @@ func (ti *TableInfo) Clone() *TableInfo {
 
 // GetIndex return the corresponding index by the given name.
 func (ti *TableInfo) GetIndex(name string) *model.IndexInfo {
+	log.Info("get index by name",
+		zap.String("name", name),
+		zap.Any("indices", ti.Indices))
 	for _, index := range ti.Indices {
 		if index != nil && index.Name.O == name {
 			return index

cdc/sink/dmlsink/mq/dispatcher/event_router.go

@@ -139,6 +139,9 @@ func (s *EventRouter) VerifyTables(infos []*model.TableInfo) error {
 				return cerror.ErrDispatcherFailed.GenWithStack(
 					"index not found when verify the table, table: %v, index: %s", table.TableName, v.IndexName)
 			}
+			// only allow the unique index to be set.
+			// For the non-unique index, if any column belongs to the index is updated,
+			// the event is not split, it may cause incorrect data consumption.
 			if !index.Unique {
 				return cerror.ErrDispatcherFailed.GenWithStack(
 					"index is not unique when verify the table, table: %v, index: %s", table.TableName, v.IndexName)