consumer(cdc): fix unstable test move_table

cdc/owner/changefeed.go

@@ -211,7 +211,7 @@ func (c *changefeed) tick(ctx cdcContext.Context, state *orchestrator.Changefeed
 		// This condition implies that the DDL resolved-ts has not yet reached checkpointTs,
 		// which implies that it would be premature to schedule tables or to update status.
 		// So we return here.
-		log.Info("barrierTs < checkpointTs, premature to schedule tables or update status",
+		log.Debug("barrierTs < checkpointTs, premature to schedule tables or update status",
 			zap.String("changefeed", c.id),
 			zap.Uint64("barrierTs", barrierTs), zap.Uint64("checkpointTs", checkpointTs))
 		return nil

cdc/processor/pipeline/sink.go

@@ -320,14 +320,14 @@ func (n *sinkNode) Receive(ctx pipeline.NodeContext) error {
 }
 
 func (n *sinkNode) HandleMessage(ctx context.Context, msg pmessage.Message) (bool, error) {
-	if n.status == TableStatusStopped {
+	if n.status.Load() == TableStatusStopped {
 		return false, cerror.ErrTableProcessorStoppedSafely.GenWithStackByArgs()
 	}
 	switch msg.Tp {
 	case pmessage.MessageTypePolymorphicEvent:
 		event := msg.PolymorphicEvent
 		if event.RawKV.OpType == model.OpTypeResolved {
-			if n.status == TableStatusInitializing {
+			if n.status.Load() == TableStatusInitializing {
 				n.status.Store(TableStatusRunning)
 			}
 			failpoint.Inject("ProcessorSyncResolvedError", func() {

cdc/sink/common/common.go

@@ -83,10 +83,8 @@ func (c *UnresolvedTxnCache) Append(filter *filter.Filter, rows ...*model.RowCha
 			// fail-fast check
 			if len(txns) != 0 && txns[len(txns)-1].commitTs > row.CommitTs {
 				log.Panic("the commitTs of the emit row is less than the received row",
-					zap.Stringer("table", row.Table),
-					zap.Uint64("emit row startTs", row.StartTs),
-					zap.Uint64("emit row commitTs", row.CommitTs),
-					zap.Uint64("last received row commitTs", txns[len(txns)-1].commitTs))
+					zap.Uint64("lastReceivedCommitTs", txns[len(txns)-1].commitTs),
+					zap.Any("row", row))
 			}
 			txns = append(txns, &txnsWithTheSameCommitTs{
 				commitTs: row.CommitTs,

cdc/sink/producer/kafka/metrics.go

@@ -225,7 +225,8 @@ func (sm *saramaMetricsMonitor) collectBrokers() {
 			"use historical brokers to collect kafka broker level metrics",
 			zap.String("changefeed", sm.changefeedID),
 			zap.Any("role", sm.role),
-			zap.Duration("duration", time.Since(start)))
+			zap.Duration("duration", time.Since(start)),
+			zap.Error(err))
 		return
 	}
 

cmd/kafka-consumer/main.go

@@ -21,6 +21,7 @@ import (
 	"net/url"
 	"os"
 	"os/signal"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -453,9 +454,36 @@ func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
 	return nil
 }
 
+type eventsGroup struct {
+	events []*model.RowChangedEvent
+}
+
+func newEventsGroup() *eventsGroup {
+	return &eventsGroup{
+		events: make([]*model.RowChangedEvent, 0),
+	}
+}
+
+func (g *eventsGroup) Append(e *model.RowChangedEvent) {
+	g.events = append(g.events, e)
+}
+
+func (g *eventsGroup) Resolve(resolveTs uint64) []*model.RowChangedEvent {
+	sort.Slice(g.events, func(i, j int) bool {
+		return g.events[i].CommitTs < g.events[j].CommitTs
+	})
+
+	i := sort.Search(len(g.events), func(i int) bool {
+		return g.events[i].CommitTs > resolveTs
+	})
+	result := g.events[:i]
+	g.events = g.events[i:]
+
+	return result
+}
+
 // ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
 func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
-	ctx := context.TODO()
 	partition := claim.Partition()
 	c.sinksMu.Lock()
 	sink := c.sinks[partition]
@@ -464,6 +492,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 		panic("sink should initialized")
 	}
 
+	eventGroups := make(map[int64]*eventsGroup)
 	for message := range claim.Messages() {
 		var (
 			decoder codec.EventBatchDecoder
@@ -547,15 +576,15 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 				if row.Table.IsPartition {
 					partitionID = row.Table.TableID
 				}
-				row.Table.TableID = c.fakeTableIDGenerator.generateFakeTableID(row.Table.Schema, row.Table.Table, partitionID)
-				err = sink.EmitRowChangedEvents(ctx, row)
-				if err != nil {
-					log.Panic("emit row changed event failed", zap.Error(err))
-				}
-				lastCommitTs, ok := sink.tablesMap.Load(row.Table.TableID)
-				if !ok || lastCommitTs.(uint64) < row.CommitTs {
-					sink.tablesMap.Store(row.Table.TableID, row.CommitTs)
+				tableID := c.fakeTableIDGenerator.generateFakeTableID(row.Table.Schema, row.Table.Table, partitionID)
+				row.Table.TableID = tableID
+
+				group, ok := eventGroups[tableID]
+				if !ok {
+					group = newEventsGroup()
+					eventGroups[tableID] = group
 				}
+				group.Append(row)
 			case model.MqMessageTypeResolved:
 				ts, err := decoder.NextResolvedEvent()
 				if err != nil {
@@ -570,7 +599,24 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 						zap.Int32("partition", partition))
 				}
 				if ts > resolvedTs {
-					log.Info("update sink resolved ts",
+					for tableID, group := range eventGroups {
+						events := group.Resolve(ts)
+						if len(events) == 0 {
+							continue
+						}
+						if err := sink.EmitRowChangedEvents(context.Background(), events...); err != nil {
+							log.Panic("emit row changed event failed",
+								zap.Any("events", events),
+								zap.Error(err),
+								zap.Int32("partition", partition))
+						}
+						commitTs := events[len(events)-1].CommitTs
+						lastCommitTs, ok := sink.tablesMap.Load(tableID)
+						if !ok || lastCommitTs.(uint64) < commitTs {
+							sink.tablesMap.Store(tableID, commitTs)
+						}
+					}
+					log.Debug("update sink resolved ts",
 						zap.Uint64("ts", ts),
 						zap.Int32("partition", partition))
 					atomic.StoreUint64(&sink.resolvedTs, ts)

tests/integration_tests/move_table/run.sh

@@ -9,10 +9,6 @@ CDC_BINARY=cdc.test
 SINK_TYPE=$1
 
 function run() {
-	# TODO: remove after kafka-consumer/main.go is fixed
-	if [ "$SINK_TYPE" == "kafka" ]; then
-		return
-	fi
 
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR