From 02be3dab2f9e9ab0ad7a3fc887ff0fa9588e97f7 Mon Sep 17 00:00:00 2001 From: nhsmw Date: Fri, 31 Jul 2026 18:14:37 +0800 Subject: [PATCH 1/7] This is an automated cherry-pick of #5824 Signed-off-by: ti-chi-bot --- cmd/kafka-consumer/writer.go | 80 +++++-------- cmd/kafka-consumer/writer_test.go | 183 +++++++++++++++++++++++++++++ cmd/pulsar-consumer/writer.go | 60 ++++------ cmd/pulsar-consumer/writer_test.go | 146 ++++++++++++++++++++++- cmd/storage-consumer/consumer.go | 4 +- cmd/util/event_group.go | 91 ++++++++------ cmd/util/event_group_test.go | 94 +++++++++++++-- 7 files changed, 518 insertions(+), 140 deletions(-) diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go index 5f9991e915..0ca062df60 100644 --- a/cmd/kafka-consumer/writer.go +++ b/cmd/kafka-consumer/writer.go @@ -609,22 +609,39 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti table = message.Table commitTs = message.GetCommitTs() ) + globalWatermark := w.globalWatermark() + if commitTs < globalWatermark { + log.Warn("DML event fallback row, since less than the global watermark, ignore it", + zap.Int64("tableID", tableID), zap.Int32("partition", progress.partition), + zap.Uint64("commitTs", commitTs), zap.Any("offset", offset), + zap.Uint64("globalWatermark", globalWatermark), + zap.Uint64("partitionWatermark", progress.watermark), + zap.Any("watermarkOffset", progress.watermarkOffset), + zap.String("schema", schema), zap.String("table", table), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) + return + } + group := progress.eventsGroup[tableID] if group == nil { group = util.NewEventsGroup(progress.partition, tableID) progress.eventsGroup[tableID] = group } + message = w.messageWithPartitionCheck(message, progress.partition, offset) + group.AppendMessage(message) if commitTs < progress.watermark { - log.Warn("DML Event fallback row, since less than the partition watermark, ignore it", + log.Warn("DML event fallback row, since less than the partition watermark, append it and sort before flush", zap.Int64("tableID", tableID), zap.Int32("partition", group.Partition), zap.Uint64("commitTs", commitTs), zap.Any("offset", offset), zap.Uint64("watermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset), - zap.String("schema", schema), zap.String("table", table)) + zap.Uint64("globalWatermark", globalWatermark), + zap.String("schema", schema), zap.String("table", table), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) return } if commitTs >= group.HighWatermark { - message = w.messageWithPartitionCheck(message, progress.partition, offset) - group.AppendMessage(message, false) log.Debug("DML event append to the group", zap.Int32("partition", group.Partition), zap.Any("offset", offset), zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark), @@ -632,54 +649,13 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti zap.Stringer("eventType", message.RowType)) return } - if w.enableTableAcrossNodes { - log.Warn("DML events fallback, but enableTableAcrossNodes is true, still append it", - zap.Int32("partition", group.Partition), zap.Any("offset", offset), - zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType)) - group.AppendMessage(w.messageWithPartitionCheck(message, progress.partition, offset), true) - return - } - switch w.protocol { - case config.ProtocolSimple: - // simple protocol set the table id for all row message, it can be known which table the row message belongs to, - // also consider the table partition. - // open protocol set the partition table id if the table is partitioned. - // for normal table, the table id is generated by the fake table id generator by using schema and table name. - // so one event group for one normal table or one table partition, replayed messages can be ignored. - log.Warn("DML event fallback row, since less than the group high watermark, ignore it", - zap.Int32("partition", progress.partition), zap.Any("offset", offset), - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType), - // zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns), - zap.Any("protocol", w.protocol)) - case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, - config.ProtocolDebezium, config.ProtocolDebeziumAvro: - // for partition table, these protocols cannot assign physical table id to each dml message, - // we cannot distinguish whether it's a real fallback event or not, still append it. - if w.partitionTableAccessor.IsPartitionTable(schema, table) { - log.Warn("DML events fallback, but the table is a partition table, still append it", - zap.Int32("partition", group.Partition), zap.Any("offset", offset), - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType), zap.Any("protocol", w.protocol)) - group.AppendMessage(w.messageWithPartitionCheck(message, progress.partition, offset), true) - return - } - log.Warn("DML event fallback row, since less than the group high watermark, ignore it", - zap.Int32("partition", progress.partition), zap.Any("offset", offset), - zap.Uint64("commitTs", commitTs), zap.Uint64("HighWatermark", group.HighWatermark), - zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType), - // zap.Any("columns", row.Columns), zap.Any("preColumns", row.PreColumns), - zap.Any("protocol", w.protocol)) - default: - log.Panic("unknown protocol", zap.Any("protocol", w.protocol)) - } + log.Warn("DML event commit ts fallback, append it and sort before flush", + zap.Int32("partition", progress.partition), zap.Any("offset", offset), + zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), + zap.Any("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset), + zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) } func openDB(ctx context.Context, dsn string) (*sql.DB, error) { diff --git a/cmd/kafka-consumer/writer_test.go b/cmd/kafka-consumer/writer_test.go index 7396b2ec32..be03a7cd9f 100644 --- a/cmd/kafka-consumer/writer_test.go +++ b/cmd/kafka-consumer/writer_test.go @@ -284,6 +284,150 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) { require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query) } +<<<<<<< HEAD +======= +func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + s := sinkmock.NewMockSink(ctrl) + flushedCommitTs := make([]uint64, 0) + s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { + flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) + event.PostFlush() + }).Times(2) + + replicaCfg := config.GetDefaultReplicaConfig() + eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false) + require.NoError(t, err) + + p := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 0, + } + w := &writer{ + progresses: []*partitionProgress{p}, + mysqlSink: s, + eventRouter: eventRouter, + protocol: config.ProtocolOpen, + } + + w.appendMessage2Group(newDMLMessageForWriterTest(20), p, kafka.Offset(1)) + w.appendMessage2Group(newDMLMessageForWriterTest(10), p, kafka.Offset(2)) + w.appendMessage2Group(newDMLMessageForWriterTest(20), p, kafka.Offset(3)) + + p.watermark = 20 + require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved)) + require.Equal(t, []uint64{10, 20}, flushedCommitTs) +} + +func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + s := sinkmock.NewMockSink(ctrl) + s.EXPECT().AddDMLEvent(gomock.Any()).Times(0) + + progress := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 20, + decoder: &singleDMLDecoder{message: newDMLMessageForWriterTest(10)}, + } + w := &writer{ + progresses: []*partitionProgress{progress}, + mysqlSink: s, + protocol: config.ProtocolOpen, + maxBatchSize: 64, + maxMessageBytes: 1, + } + + needCommit := w.WriteMessage(ctx, &kafka.Message{ + TopicPartition: kafka.TopicPartition{Partition: 0, Offset: kafka.Offset(10)}, + }) + + require.False(t, needCommit) + require.Nil(t, progress.eventsGroup[1]) +} + +func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) { + replicaCfg := config.GetDefaultReplicaConfig() + eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false) + require.NoError(t, err) + + progress := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 20, + } + w := &writer{ + progresses: []*partitionProgress{ + progress, + {partition: 1, watermark: 5}, + }, + eventRouter: eventRouter, + protocol: config.ProtocolOpen, + } + + w.appendMessage2Group(newDMLMessageForWriterTest(10), progress, kafka.Offset(10)) + + require.NotNil(t, progress.eventsGroup[1]) + resolved := progress.eventsGroup[1].ResolveInto(20, nil) + require.Len(t, resolved, 1) + require.Equal(t, uint64(10), resolved[0].GetCommitTs()) +} + +func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) { + replicaCfg := config.GetDefaultReplicaConfig() + eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, true) + require.NoError(t, err) + + w := &writer{ + progresses: []*partitionProgress{{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}}, + eventRouter: eventRouter, + protocol: config.ProtocolAvro, + partitionTableAccessor: codeccommon.NewPartitionTableAccessor(), + } + + ddl := &commonEvent.DDLEvent{ + Query: "CREATE TABLE `target`.`dst` LIKE `target`.`src`", + SchemaName: "source", + TableName: "dst", + Type: byte(timodel.ActionCreateTable), + TableInfo: &common.TableInfo{ + TableName: common.TableName{ + Schema: "source", + Table: "dst", + IsPartition: true, + TargetSchema: "target", + TargetTable: "dst", + }, + }, + } + w.onDDL(ddl) + require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst")) + + newDMLEvent := func(commitTs uint64) *commonEvent.DMLEvent { + return &commonEvent.DMLEvent{ + PhysicalTableID: 1, + CommitTs: commitTs, + RowTypes: []common.RowType{common.RowTypeUpdate}, + Rows: chunk.NewChunkWithCapacity(nil, 0), + TableInfo: &common.TableInfo{ + TableName: common.TableName{Schema: "target", Table: "dst"}, + }, + } + } + + progress := w.progresses[0] + w.appendMessage2Group(codeccommon.NewDMLMessageFromEvent(newDMLEvent(200)), progress, kafka.Offset(10)) + w.appendMessage2Group(codeccommon.NewDMLMessageFromEvent(newDMLEvent(100)), progress, kafka.Offset(11)) + + resolved := progress.eventsGroup[1].ResolveInto(150, nil) + require.Len(t, resolved, 1) + require.Equal(t, uint64(100), resolved[0].GetCommitTs()) +} + +>>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824)) func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) { for _, protocol := range []config.Protocol{ config.ProtocolDebezium, @@ -333,3 +477,42 @@ func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) { }) } } + +func newDMLMessageForWriterTest(commitTs uint64) *codeccommon.DMLMessage { + return codeccommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeUpdate, func() *commonEvent.DMLEvent { + return &commonEvent.DMLEvent{ + PhysicalTableID: 1, + CommitTs: commitTs, + RowTypes: []common.RowType{common.RowTypeUpdate}, + Rows: chunk.NewChunkWithCapacity(nil, 0), + TableInfo: &common.TableInfo{ + TableName: common.TableName{Schema: "test", Table: "t", TableID: 1}, + }, + } + }) +} + +type singleDMLDecoder struct { + message *codeccommon.DMLMessage + consumed bool +} + +func (d *singleDMLDecoder) AddKeyValue(_, _ []byte) { +} + +func (d *singleDMLDecoder) HasNext() (codeccommon.MessageType, bool) { + return codeccommon.MessageTypeRow, !d.consumed +} + +func (d *singleDMLDecoder) NextResolvedEvent() uint64 { + return 0 +} + +func (d *singleDMLDecoder) NextDMLMessage() *codeccommon.DMLMessage { + d.consumed = true + return d.message +} + +func (d *singleDMLDecoder) NextDDLEvent() *commonEvent.DDLEvent { + return nil +} diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go index 418fc5fc35..1802e27f90 100644 --- a/cmd/pulsar-consumer/writer.go +++ b/cmd/pulsar-consumer/writer.go @@ -498,55 +498,47 @@ func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *parti table = message.Table commitTs = message.GetCommitTs() ) + globalWatermark := w.globalWatermark() + if commitTs < globalWatermark { + log.Warn("DML event fallback row, since less than the global watermark, ignore it", + zap.Int64("tableID", tableID), zap.Int32("partition", progress.partition), + zap.Uint64("commitTs", commitTs), + zap.Uint64("globalWatermark", globalWatermark), + zap.Uint64("partitionWatermark", progress.watermark), + zap.String("schema", schema), zap.String("table", table), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) + return + } + group := progress.eventsGroup[tableID] if group == nil { group = util.NewEventsGroup(progress.partition, tableID) progress.eventsGroup[tableID] = group } + group.AppendMessage(message) if commitTs < progress.watermark { - log.Warn("DML Event fallback row, since less than the partition watermark, ignore it", + log.Warn("DML event fallback row, since less than the partition watermark, append it and sort before flush", zap.Int64("tableID", tableID), zap.Int32("partition", group.Partition), zap.Uint64("commitTs", commitTs), zap.Uint64("watermark", progress.watermark), - zap.String("schema", schema), zap.String("table", table)) + zap.Uint64("globalWatermark", globalWatermark), + zap.String("schema", schema), zap.String("table", table), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) return } if commitTs >= group.HighWatermark { - group.AppendMessage(message, false) log.Debug("DML event append to the group", zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType)) return } - if w.enableTableAcrossNodes { - log.Warn("DML events fallback, but enableTableAcrossNodes is true, still append it", - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType)) - group.AppendMessage(message, true) - return - } - switch w.protocol { - case config.ProtocolCanalJSON: - // for partition table, the canal-json message cannot assign physical table id to each dml message, - // we cannot distinguish whether it's a real fallback event or not, still append it. - isPartitionTable := w.partitionTableAccessor != nil && - w.partitionTableAccessor.IsPartitionTable(schema, table) - if isPartitionTable { - log.Warn("DML events fallback, but it's canal-json and partition table, still append it", - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType)) - group.AppendMessage(message, true) - return - } - log.Warn("DML event fallback row, since less than the group high watermark, ignore it", - zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), - zap.Any("partitionWatermark", progress.watermark), zap.Any("watermark", progress.watermark), - zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), - zap.Stringer("eventType", message.RowType), - zap.Any("protocol", w.protocol), zap.Bool("IsPartition", isPartitionTable)) - default: - log.Panic("unknown protocol", zap.Any("protocol", w.protocol)) - } + log.Warn("DML event commit ts fallback, append it and sort before flush", + zap.Int32("partition", progress.partition), + zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), + zap.Any("partitionWatermark", progress.watermark), + zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), + zap.Stringer("eventType", message.RowType), + zap.Any("protocol", w.protocol), zap.Bool("enableTableAcrossNodes", w.enableTableAcrossNodes)) } diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index f071376b97..e28d90ccc4 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -28,6 +28,7 @@ import ( "github.com/pingcap/ticdc/pkg/config" codeccommon "github.com/pingcap/ticdc/pkg/sink/codec/common" timodel "github.com/pingcap/tidb/pkg/meta/model" + "github.com/pingcap/tidb/pkg/util/chunk" "github.com/stretchr/testify/require" ) @@ -285,6 +286,135 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) { require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query) } +<<<<<<< HEAD +======= +func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + s := sinkmock.NewMockSink(ctrl) + flushedCommitTs := make([]uint64, 0) + s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { + flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) + event.PostFlush() + }).Times(2) + + p := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 0, + } + w := &writer{ + progresses: []*partitionProgress{p}, + mysqlSink: s, + protocol: config.ProtocolCanalJSON, + } + + w.appendMessage2Group(newDMLMessageForWriterTest(20), p) + w.appendMessage2Group(newDMLMessageForWriterTest(10), p) + w.appendMessage2Group(newDMLMessageForWriterTest(20), p) + + p.watermark = 20 + require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved)) + require.Equal(t, []uint64{10, 20}, flushedCommitTs) +} + +func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + s := sinkmock.NewMockSink(ctrl) + s.EXPECT().AddDMLEvent(gomock.Any()).Times(0) + + decoder := &deferredDMLDecoder{ + row: &commonEvent.DMLEvent{ + PhysicalTableID: 1, + CommitTs: 10, + RowTypes: []common.RowType{common.RowTypeInsert}, + TableInfo: &common.TableInfo{ + TableName: common.TableName{Schema: "test", Table: "t", TableID: 1}, + }, + }, + } + progress := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 20, + decoder: decoder, + } + w := &writer{ + progresses: []*partitionProgress{progress}, + mysqlSink: s, + protocol: config.ProtocolCanalJSON, + } + + needCommit := w.WriteMessage(ctx, fakePulsarMessage{key: "k", payload: []byte(`{"fake":"row"}`)}) + + require.False(t, needCommit) + require.Nil(t, progress.eventsGroup[1]) +} + +func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) { + progress := &partitionProgress{ + partition: 0, + eventsGroup: make(map[int64]*util.EventsGroup), + watermark: 20, + } + w := &writer{ + progresses: []*partitionProgress{ + progress, + {partition: 1, watermark: 5}, + }, + protocol: config.ProtocolCanalJSON, + } + + w.appendMessage2Group(newDMLMessageForWriterTest(10), progress) + + require.NotNil(t, progress.eventsGroup[1]) + resolved := progress.eventsGroup[1].ResolveInto(20, nil) + require.Len(t, resolved, 1) + require.Equal(t, uint64(10), resolved[0].GetCommitTs()) +} + +func TestOnDDLMarksRoutedCreateTableLikePartitionTable(t *testing.T) { + w := &writer{ + progresses: []*partitionProgress{ + {partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}, + }, + protocol: config.ProtocolCanalJSON, + partitionTableAccessor: codeccommon.NewPartitionTableAccessor(), + } + + ddl := &commonEvent.DDLEvent{ + Query: "CREATE TABLE `target`.`dst` LIKE `target`.`src`", + SchemaName: "source", + TableName: "dst", + Type: byte(timodel.ActionCreateTable), + TableInfo: &common.TableInfo{ + TableName: common.TableName{ + Schema: "source", + Table: "dst", + IsPartition: true, + TargetSchema: "target", + TargetTable: "dst", + }, + }, + } + w.onDDL(ddl) + require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst")) + + newDMLMessage := func(commitTs uint64) *codeccommon.DMLMessage { + return codeccommon.NewDMLMessage(1, "target", "dst", commitTs, common.RowTypeUpdate, nil) + } + + progress := w.progresses[0] + w.appendMessage2Group(newDMLMessage(200), progress) + w.appendMessage2Group(newDMLMessage(100), progress) + + resolved := progress.eventsGroup[1].ResolveInto(150, nil) + require.Len(t, resolved, 1) + require.Equal(t, uint64(100), resolved[0].GetCommitTs()) +} + +>>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824)) func TestWriteMessageDefersDMLAssemblyUntilFlush(t *testing.T) { ctx := context.Background() ctrl := gomock.NewController(t) @@ -358,7 +488,7 @@ func (d *deferredDMLDecoder) NextResolvedEvent() uint64 { func (d *deferredDMLDecoder) NextDMLMessage() *codeccommon.DMLMessage { d.nextDMLMessageCount++ - return codeccommon.NewDMLMessage(1, "test", "t", 100, common.RowTypeInsert, func() *commonEvent.DMLEvent { + return codeccommon.NewDMLMessage(1, "test", "t", d.row.CommitTs, common.RowTypeInsert, func() *commonEvent.DMLEvent { d.toDMLEventCount++ return d.row }) @@ -368,6 +498,20 @@ func (d *deferredDMLDecoder) NextDDLEvent() *commonEvent.DDLEvent { return nil } +func newDMLMessageForWriterTest(commitTs uint64) *codeccommon.DMLMessage { + return codeccommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeUpdate, func() *commonEvent.DMLEvent { + return &commonEvent.DMLEvent{ + PhysicalTableID: 1, + CommitTs: commitTs, + RowTypes: []common.RowType{common.RowTypeUpdate}, + Rows: chunk.NewChunkWithCapacity(nil, 0), + TableInfo: &common.TableInfo{ + TableName: common.TableName{Schema: "test", Table: "t", TableID: 1}, + }, + } + }) +} + type fakePulsarMessage struct { key string payload []byte diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 7e3ec687d0..42081fbc9d 100644 --- a/cmd/storage-consumer/consumer.go +++ b/cmd/storage-consumer/consumer.go @@ -264,7 +264,7 @@ func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAc c.eventsGroup[tableID] = group } if commitTs >= group.HighWatermark { - group.AppendMessage(message, false) + group.AppendMessage(message) log.Debug("DML event append to the group", zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), @@ -276,7 +276,7 @@ func (c *consumer) appendMessage2Group(message *common.DMLMessage, enableTableAc zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark), zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID), zap.Stringer("eventType", message.RowType)) - group.AppendMessage(message, true) + group.AppendMessage(message) return } log.Warn("dml event commit ts fallback, ignore", diff --git a/cmd/util/event_group.go b/cmd/util/event_group.go index b6eebe2039..2056212408 100644 --- a/cmd/util/event_group.go +++ b/cmd/util/event_group.go @@ -14,6 +14,7 @@ package util import ( + "math" "sort" "github.com/pingcap/log" @@ -48,65 +49,79 @@ func NewEventsGroup(partition int32, tableID int64) *EventsGroup { } // AppendMessage appends a message to event groups. -func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage, force bool) { +func (g *EventsGroup) AppendMessage(message *codeccommon.DMLMessage) { commitTs := message.GetCommitTs() if commitTs > g.HighWatermark { g.HighWatermark = commitTs } + g.messages = append(g.messages, message) +} - var lastMessage *codeccommon.DMLMessage - if len(g.messages) > 0 { - lastMessage = g.messages[len(g.messages)-1] +// ResolveInto appends all messages with CommitTs <= resolve into dst in commit-ts order and removes +// them from the group. ResolveInto copies pointers into dst first, then clears the resolved messages +// so Go GC can reclaim them once downstream is done with them. +func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) []*codeccommon.DMLMessage { + if len(g.messages) == 0 { + return dst } - if lastMessage == nil || lastMessage.GetCommitTs() <= commitTs { - g.messages = append(g.messages, message) - return + original := g.messages + remaining := g.messages[:0] + resolved := make([]*codeccommon.DMLMessage, 0, len(g.messages)) + + var ( + lastCommitTs uint64 + outOfOrder bool + outOfOrderLastTs uint64 + outOfOrderCommitTs uint64 + ) + for _, message := range g.messages { + commitTs := message.GetCommitTs() + if commitTs > resolve { + remaining = append(remaining, message) + continue + } + if len(resolved) > 0 && commitTs < lastCommitTs && !outOfOrder { + outOfOrder = true + outOfOrderLastTs = lastCommitTs + outOfOrderCommitTs = commitTs + } + lastCommitTs = commitTs + resolved = append(resolved, message) } - - if force { - i := sort.Search(len(g.messages), func(i int) bool { - return g.messages[i].GetCommitTs() > commitTs - }) - g.messages = append(g.messages, nil) - copy(g.messages[i+1:], g.messages[i:]) - g.messages[i] = message - return + if len(resolved) == 0 { + return dst } - log.Panic("append event with smaller commit ts", - zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), - zap.Uint64("lastCommitTs", lastMessage.GetCommitTs()), zap.Uint64("commitTs", commitTs)) -} -// ResolveInto appends all messages with CommitTs <= resolve into dst and removes them from the group. -// ResolveInto copies pointers into dst first, then clears the resolved prefix so Go GC can reclaim -// resolved messages once downstream is done with them. -func (g *EventsGroup) ResolveInto(resolve uint64, dst []*codeccommon.DMLMessage) []*codeccommon.DMLMessage { - i := sort.Search(len(g.messages), func(i int) bool { - return g.messages[i].GetCommitTs() > resolve - }) - if i == 0 { - return dst + if outOfOrder { + log.Warn("DML events are out of order before flush, sort them", + zap.Int32("partition", g.Partition), + zap.Int64("tableID", g.tableID), + zap.Uint64("resolveTs", resolve), + zap.Int("resolved", len(resolved)), + zap.Uint64("lastCommitTs", outOfOrderLastTs), + zap.Uint64("commitTs", outOfOrderCommitTs)) + sort.SliceStable(resolved, func(i, j int) bool { + return resolved[i].GetCommitTs() < resolved[j].GetCommitTs() + }) } - // Copy pointers out first so we can safely clear the group's slice without affecting callers. - dst = append(dst, g.messages[:i]...) - clear(g.messages[:i]) - g.messages = g.messages[i:] + dst = append(dst, resolved...) + clear(original[len(remaining):]) + g.messages = remaining if len(g.messages) != 0 { + firstCommitTs := g.messages[0].GetCommitTs() log.Debug("not all events resolved", zap.Int32("partition", g.Partition), zap.Int64("tableID", g.tableID), - zap.Int("resolved", i), zap.Int("remained", len(g.messages)), - zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", g.messages[0].GetCommitTs())) + zap.Int("resolved", len(resolved)), zap.Int("remained", len(g.messages)), + zap.Uint64("resolveTs", resolve), zap.Uint64("firstCommitTs", firstCommitTs)) } return dst } // GetAllMessages gets all messages. func (g *EventsGroup) GetAllMessages() []*codeccommon.DMLMessage { - result := g.messages - g.messages = nil - return result + return g.ResolveInto(math.MaxUint64, nil) } // AppendOrMergeDMLEvent appends a DML event, or merges it into the previous event diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index 0bb4f82fe9..9ec6e79bff 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -37,25 +37,32 @@ func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.D } } +<<<<<<< HEAD func TestEventsGroupResolveIntoAppendsAndClearsResolvedPrefix(t *testing.T) { // Scenario: A consumer resolves a prefix of events by watermark/commit-ts and appends them // array to avoid retaining already-flushed events and causing unbounded memory growth. +======= +func TestEventsGroupResolveIntoAppendsAndClearsResolvedMessages(t *testing.T) { + // Scenario: A consumer resolves events by watermark/commit-ts and appends them into a downstream + // batch slice. We must clear resolved messages in the group's backing array to avoid retaining + // already-flushed events and causing unbounded memory growth. +>>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824)) // // Steps: // 1. Append 3 events with increasing CommitTs. // 2. Call ResolveInto with resolve=2 and a nil dst. // 3. Verify (a) returned events are correct, (b) group keeps only the remaining event, - // (c) the resolved prefix in the original backing slice is cleared (nil'd). + // (c) resolved messages in the original backing slice are cleared (nil'd). group := NewEventsGroup(0, 1) m1 := newTestDMLMessage(1) m2 := newTestDMLMessage(2) m3 := newTestDMLMessage(3) - group.AppendMessage(m1, false) - group.AppendMessage(m2, false) - group.AppendMessage(m3, false) + group.AppendMessage(m1) + group.AppendMessage(m2) + group.AppendMessage(m3) // Keep a reference to the original slice header so we can validate that ResolveInto clears - // the resolved prefix in-place (this is what prevents GC retention of flushed events). + // resolved messages in-place (this is what prevents GC retention of flushed events). original := group.messages var dst []*codeccommon.DMLMessage @@ -68,11 +75,11 @@ func TestEventsGroupResolveIntoAppendsAndClearsResolvedPrefix(t *testing.T) { require.Len(t, group.messages, 1) require.Same(t, m3, group.messages[0]) - // The resolved prefix must be nil so the group doesn't keep flushed events alive via its - // backing array (classic Go slice memory retention pitfall). - require.Nil(t, original[0]) + // The unresolved event is compacted to the front, and the tail is cleared so the group + // doesn't keep flushed events alive via its backing array. + require.Same(t, m3, original[0]) require.Nil(t, original[1]) - require.Same(t, m3, original[2]) + require.Nil(t, original[2]) } func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) { @@ -81,8 +88,8 @@ func TestEventsGroupResolveIntoNoopWhenNothingResolved(t *testing.T) { group := NewEventsGroup(0, 1) m1 := newTestDMLMessage(10) m2 := newTestDMLMessage(20) - group.AppendMessage(m1, false) - group.AppendMessage(m2, false) + group.AppendMessage(m1) + group.AppendMessage(m2) original := group.messages dst := make([]*codeccommon.DMLMessage, 0, 1) @@ -104,8 +111,8 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) { group := NewEventsGroup(0, 1) m1 := newTestDMLMessage(1) m2 := newTestDMLMessage(2) - group.AppendMessage(m1, false) - group.AppendMessage(m2, false) + group.AppendMessage(m1) + group.AppendMessage(m2) original := group.messages var dst []*codeccommon.DMLMessage @@ -120,6 +127,67 @@ func TestEventsGroupResolveIntoClearsAllWhenFullyResolved(t *testing.T) { require.Nil(t, original[1]) } +func TestEventsGroupResolveIntoSortsOutOfOrderResolvedMessages(t *testing.T) { + group := NewEventsGroup(0, 1) + m1 := newTestDMLMessage(20) + m2 := newTestDMLMessage(10) + m3 := newTestDMLMessage(30) + group.AppendMessage(m1) + group.AppendMessage(m2) + group.AppendMessage(m3) + + original := group.messages + var dst []*codeccommon.DMLMessage + dst = group.ResolveInto(25, dst) + + require.Len(t, dst, 2) + require.Same(t, m2, dst[0]) + require.Same(t, m1, dst[1]) + + require.Len(t, group.messages, 1) + require.Same(t, m3, group.messages[0]) + require.Same(t, m3, original[0]) + require.Nil(t, original[1]) + require.Nil(t, original[2]) +} + +func TestEventsGroupResolveIntoKeepsSameCommitTsStable(t *testing.T) { + group := NewEventsGroup(0, 1) + m1 := newTestDMLMessage(20) + m2 := newTestDMLMessage(10) + m3 := newTestDMLMessage(20) + group.AppendMessage(m1) + group.AppendMessage(m2) + group.AppendMessage(m3) + + var dst []*codeccommon.DMLMessage + dst = group.ResolveInto(20, dst) + + require.Len(t, dst, 3) + require.Same(t, m2, dst[0]) + require.Same(t, m1, dst[1]) + require.Same(t, m3, dst[2]) + require.Empty(t, group.messages) +} + +func TestEventsGroupGetAllMessagesSortsOutOfOrderMessages(t *testing.T) { + group := NewEventsGroup(0, 1) + m1 := newTestDMLMessage(20) + m2 := newTestDMLMessage(10) + m3 := newTestDMLMessage(30) + group.AppendMessage(m1) + group.AppendMessage(m2) + group.AppendMessage(m3) + + messages := group.GetAllMessages() + + require.Len(t, messages, 3) + require.Same(t, m2, messages[0]) + require.Same(t, m1, messages[1]) + require.Same(t, m3, messages[2]) + require.Empty(t, group.messages) +} + func TestAppendOrMergeDMLEventMergesSameCommitTs(t *testing.T) { var flushed []int e1 := newTestDMLEvent(10, common.RowTypeInsert) From c1014ffa1e0633ab85cadc3c2072409f8cf5ce6b Mon Sep 17 00:00:00 2001 From: nhsmw Date: Fri, 31 Jul 2026 23:05:07 +0800 Subject: [PATCH 2/7] Update event_group_test.go --- cmd/util/event_group_test.go | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index 9ec6e79bff..c8d3d8770f 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -37,16 +37,11 @@ func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.D } } -<<<<<<< HEAD -func TestEventsGroupResolveIntoAppendsAndClearsResolvedPrefix(t *testing.T) { - // Scenario: A consumer resolves a prefix of events by watermark/commit-ts and appends them - // array to avoid retaining already-flushed events and causing unbounded memory growth. -======= + func TestEventsGroupResolveIntoAppendsAndClearsResolvedMessages(t *testing.T) { // Scenario: A consumer resolves events by watermark/commit-ts and appends them into a downstream // batch slice. We must clear resolved messages in the group's backing array to avoid retaining // already-flushed events and causing unbounded memory growth. ->>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824)) // // Steps: // 1. Append 3 events with increasing CommitTs. From ecf3ea546b385aed0423021c82f784f5303940b2 Mon Sep 17 00:00:00 2001 From: nhsmw Date: Fri, 31 Jul 2026 23:06:39 +0800 Subject: [PATCH 3/7] Update writer_test.go --- cmd/pulsar-consumer/writer_test.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index e28d90ccc4..2128c2f0ae 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -286,8 +286,6 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) { require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query) } -<<<<<<< HEAD -======= func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { ctx := context.Background() ctrl := gomock.NewController(t) @@ -414,7 +412,6 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTable(t *testing.T) { require.Equal(t, uint64(100), resolved[0].GetCommitTs()) } ->>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824)) func TestWriteMessageDefersDMLAssemblyUntilFlush(t *testing.T) { ctx := context.Background() ctrl := gomock.NewController(t) From f8a45fc57102ad2c7d1edf0cec0b5e9b560f63e1 Mon Sep 17 00:00:00 2001 From: nhsmw Date: Fri, 31 Jul 2026 23:07:16 +0800 Subject: [PATCH 4/7] Update writer_test.go --- cmd/kafka-consumer/writer_test.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/cmd/kafka-consumer/writer_test.go b/cmd/kafka-consumer/writer_test.go index be03a7cd9f..023e1bc6f9 100644 --- a/cmd/kafka-consumer/writer_test.go +++ b/cmd/kafka-consumer/writer_test.go @@ -284,8 +284,6 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) { require.Equal(t, "CREATE TABLE `common_1`.`a` (`a` BIGINT PRIMARY KEY,`b` INT)", w.ddlList[0].Query) } -<<<<<<< HEAD -======= func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { ctx := context.Background() ctrl := gomock.NewController(t) @@ -427,7 +425,6 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) { require.Equal(t, uint64(100), resolved[0].GetCommitTs()) } ->>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824)) func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) { for _, protocol := range []config.Protocol{ config.ProtocolDebezium, From 505158374f8b106bd00a0254c89096bd3366ea89 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 31 Jul 2026 15:27:23 +0000 Subject: [PATCH 5/7] fmt Signed-off-by: wk989898 --- cmd/util/event_group_test.go | 1 - 1 file changed, 1 deletion(-) diff --git a/cmd/util/event_group_test.go b/cmd/util/event_group_test.go index c8d3d8770f..da2cbec9c4 100644 --- a/cmd/util/event_group_test.go +++ b/cmd/util/event_group_test.go @@ -37,7 +37,6 @@ func newTestDMLEvent(commitTs uint64, rowTypes ...common.RowType) *commonEvent.D } } - func TestEventsGroupResolveIntoAppendsAndClearsResolvedMessages(t *testing.T) { // Scenario: A consumer resolves events by watermark/commit-ts and appends them into a downstream // batch slice. We must clear resolved messages in the group's backing array to avoid retaining From f341a907e8f0ebf674984d0a5cceb3057872fd06 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 31 Jul 2026 16:26:41 +0000 Subject: [PATCH 6/7] update Signed-off-by: wk989898 --- cmd/kafka-consumer/writer_test.go | 34 +++++++++++++++--------------- cmd/pulsar-consumer/writer_test.go | 8 +++---- 2 files changed, 20 insertions(+), 22 deletions(-) diff --git a/cmd/kafka-consumer/writer_test.go b/cmd/kafka-consumer/writer_test.go index 023e1bc6f9..b2fcb8009f 100644 --- a/cmd/kafka-consumer/writer_test.go +++ b/cmd/kafka-consumer/writer_test.go @@ -18,9 +18,11 @@ import ( "testing" "github.com/confluentinc/confluent-kafka-go/v2/kafka" + "github.com/golang/mock/gomock" "github.com/pingcap/ticdc/cmd/util" "github.com/pingcap/ticdc/downstreamadapter/sink" "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" + "github.com/pingcap/ticdc/downstreamadapter/sink/mock" "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" @@ -287,7 +289,7 @@ func TestWriterWrite_handlesOutOfOrderDDLsByCommitTs(t *testing.T) { func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { ctx := context.Background() ctrl := gomock.NewController(t) - s := sinkmock.NewMockSink(ctrl) + s := mock.NewMockSink(ctrl) flushedCommitTs := make([]uint64, 0) s.EXPECT().AddDMLEvent(gomock.Any()).Do(func(event *commonEvent.DMLEvent) { flushedCommitTs = append(flushedCommitTs, event.GetCommitTs()) @@ -315,14 +317,14 @@ func TestWriterWrite_sortsOutOfOrderDMLByWatermark(t *testing.T) { w.appendMessage2Group(newDMLMessageForWriterTest(20), p, kafka.Offset(3)) p.watermark = 20 - require.True(t, w.Write(ctx, codeccommon.MessageTypeResolved)) + require.True(t, w.Write(ctx, codecCommon.MessageTypeResolved)) require.Equal(t, []uint64{10, 20}, flushedCommitTs) } func TestWriteMessageIgnoresFallbackDMLBelowGlobalWatermark(t *testing.T) { ctx := context.Background() ctrl := gomock.NewController(t) - s := sinkmock.NewMockSink(ctrl) + s := mock.NewMockSink(ctrl) s.EXPECT().AddDMLEvent(gomock.Any()).Times(0) progress := &partitionProgress{ @@ -383,7 +385,7 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) { progresses: []*partitionProgress{{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}}, eventRouter: eventRouter, protocol: config.ProtocolAvro, - partitionTableAccessor: codeccommon.NewPartitionTableAccessor(), + partitionTableAccessor: codecCommon.NewPartitionTableAccessor(), } ddl := &commonEvent.DDLEvent{ @@ -393,11 +395,9 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) { Type: byte(timodel.ActionCreateTable), TableInfo: &common.TableInfo{ TableName: common.TableName{ - Schema: "source", - Table: "dst", - IsPartition: true, - TargetSchema: "target", - TargetTable: "dst", + Schema: "source", + Table: "dst", + IsPartition: true, }, }, } @@ -417,8 +417,8 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) { } progress := w.progresses[0] - w.appendMessage2Group(codeccommon.NewDMLMessageFromEvent(newDMLEvent(200)), progress, kafka.Offset(10)) - w.appendMessage2Group(codeccommon.NewDMLMessageFromEvent(newDMLEvent(100)), progress, kafka.Offset(11)) + w.appendMessage2Group(codecCommon.NewDMLMessageFromEvent(newDMLEvent(200)), progress, kafka.Offset(10)) + w.appendMessage2Group(codecCommon.NewDMLMessageFromEvent(newDMLEvent(100)), progress, kafka.Offset(11)) resolved := progress.eventsGroup[1].ResolveInto(150, nil) require.Len(t, resolved, 1) @@ -475,8 +475,8 @@ func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) { } } -func newDMLMessageForWriterTest(commitTs uint64) *codeccommon.DMLMessage { - return codeccommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeUpdate, func() *commonEvent.DMLEvent { +func newDMLMessageForWriterTest(commitTs uint64) *codecCommon.DMLMessage { + return codecCommon.NewDMLMessage(1, "test", "t", commitTs, common.RowTypeUpdate, func() *commonEvent.DMLEvent { return &commonEvent.DMLEvent{ PhysicalTableID: 1, CommitTs: commitTs, @@ -490,22 +490,22 @@ func newDMLMessageForWriterTest(commitTs uint64) *codeccommon.DMLMessage { } type singleDMLDecoder struct { - message *codeccommon.DMLMessage + message *codecCommon.DMLMessage consumed bool } func (d *singleDMLDecoder) AddKeyValue(_, _ []byte) { } -func (d *singleDMLDecoder) HasNext() (codeccommon.MessageType, bool) { - return codeccommon.MessageTypeRow, !d.consumed +func (d *singleDMLDecoder) HasNext() (codecCommon.MessageType, bool) { + return codecCommon.MessageTypeRow, !d.consumed } func (d *singleDMLDecoder) NextResolvedEvent() uint64 { return 0 } -func (d *singleDMLDecoder) NextDMLMessage() *codeccommon.DMLMessage { +func (d *singleDMLDecoder) NextDMLMessage() *codecCommon.DMLMessage { d.consumed = true return d.message } diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index 2128c2f0ae..93f2678d26 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -388,11 +388,9 @@ func TestOnDDLMarksRoutedCreateTableLikePartitionTable(t *testing.T) { Type: byte(timodel.ActionCreateTable), TableInfo: &common.TableInfo{ TableName: common.TableName{ - Schema: "source", - Table: "dst", - IsPartition: true, - TargetSchema: "target", - TargetTable: "dst", + Schema: "source", + Table: "dst", + IsPartition: true, }, }, } From 250f25f14ac92fd9c77b0801f356ea0581256b8d Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 7 Aug 2026 08:58:03 +0000 Subject: [PATCH 7/7] update Signed-off-by: wk989898 --- cmd/kafka-consumer/writer.go | 1 - cmd/kafka-consumer/writer_test.go | 49 ------------------------------ cmd/pulsar-consumer/writer.go | 1 - cmd/pulsar-consumer/writer_test.go | 38 ----------------------- 4 files changed, 89 deletions(-) diff --git a/cmd/kafka-consumer/writer.go b/cmd/kafka-consumer/writer.go index 0ca062df60..6ac5e0e1a4 100644 --- a/cmd/kafka-consumer/writer.go +++ b/cmd/kafka-consumer/writer.go @@ -554,7 +554,6 @@ func (w *writer) markPartitionTableFromDDL(ddl *event.DDLEvent) bool { w.addPartitionTable(ddl.GetSchemaName(), ddl.GetTableName()) w.addPartitionTable(ddl.TableInfo.GetSchemaName(), ddl.TableInfo.GetTableName()) - w.addPartitionTable(ddl.TableInfo.GetSchemaName(), ddl.TableInfo.GetTableName()) return true } diff --git a/cmd/kafka-consumer/writer_test.go b/cmd/kafka-consumer/writer_test.go index b2fcb8009f..b22c861d71 100644 --- a/cmd/kafka-consumer/writer_test.go +++ b/cmd/kafka-consumer/writer_test.go @@ -376,55 +376,6 @@ func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) { require.Equal(t, uint64(10), resolved[0].GetCommitTs()) } -func TestOnDDLMarksRoutedCreateTableLikePartitionTableForAvro(t *testing.T) { - replicaCfg := config.GetDefaultReplicaConfig() - eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, true) - require.NoError(t, err) - - w := &writer{ - progresses: []*partitionProgress{{partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}}, - eventRouter: eventRouter, - protocol: config.ProtocolAvro, - partitionTableAccessor: codecCommon.NewPartitionTableAccessor(), - } - - ddl := &commonEvent.DDLEvent{ - Query: "CREATE TABLE `target`.`dst` LIKE `target`.`src`", - SchemaName: "source", - TableName: "dst", - Type: byte(timodel.ActionCreateTable), - TableInfo: &common.TableInfo{ - TableName: common.TableName{ - Schema: "source", - Table: "dst", - IsPartition: true, - }, - }, - } - w.onDDL(ddl) - require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst")) - - newDMLEvent := func(commitTs uint64) *commonEvent.DMLEvent { - return &commonEvent.DMLEvent{ - PhysicalTableID: 1, - CommitTs: commitTs, - RowTypes: []common.RowType{common.RowTypeUpdate}, - Rows: chunk.NewChunkWithCapacity(nil, 0), - TableInfo: &common.TableInfo{ - TableName: common.TableName{Schema: "target", Table: "dst"}, - }, - } - } - - progress := w.progresses[0] - w.appendMessage2Group(codecCommon.NewDMLMessageFromEvent(newDMLEvent(200)), progress, kafka.Offset(10)) - w.appendMessage2Group(codecCommon.NewDMLMessageFromEvent(newDMLEvent(100)), progress, kafka.Offset(11)) - - resolved := progress.eventsGroup[1].ResolveInto(150, nil) - require.Len(t, resolved, 1) - require.Equal(t, uint64(100), resolved[0].GetCommitTs()) -} - func TestAppendRow2GroupKeepsDebeziumPartitionTableFallback(t *testing.T) { for _, protocol := range []config.Protocol{ config.ProtocolDebezium, diff --git a/cmd/pulsar-consumer/writer.go b/cmd/pulsar-consumer/writer.go index 1802e27f90..74abf91651 100644 --- a/cmd/pulsar-consumer/writer.go +++ b/cmd/pulsar-consumer/writer.go @@ -480,7 +480,6 @@ func (w *writer) markPartitionTableFromDDL(ddl *commonEvent.DDLEvent) bool { w.addPartitionTable(ddl.GetSchemaName(), ddl.GetTableName()) w.addPartitionTable(ddl.TableInfo.GetSchemaName(), ddl.TableInfo.GetTableName()) - w.addPartitionTable(ddl.TableInfo.GetSchemaName(), ddl.TableInfo.GetTableName()) return true } diff --git a/cmd/pulsar-consumer/writer_test.go b/cmd/pulsar-consumer/writer_test.go index 93f2678d26..a3314153ce 100644 --- a/cmd/pulsar-consumer/writer_test.go +++ b/cmd/pulsar-consumer/writer_test.go @@ -372,44 +372,6 @@ func TestAppendMessageKeepsFallbackDMLAboveGlobalWatermark(t *testing.T) { require.Equal(t, uint64(10), resolved[0].GetCommitTs()) } -func TestOnDDLMarksRoutedCreateTableLikePartitionTable(t *testing.T) { - w := &writer{ - progresses: []*partitionProgress{ - {partition: 0, eventsGroup: make(map[int64]*util.EventsGroup)}, - }, - protocol: config.ProtocolCanalJSON, - partitionTableAccessor: codeccommon.NewPartitionTableAccessor(), - } - - ddl := &commonEvent.DDLEvent{ - Query: "CREATE TABLE `target`.`dst` LIKE `target`.`src`", - SchemaName: "source", - TableName: "dst", - Type: byte(timodel.ActionCreateTable), - TableInfo: &common.TableInfo{ - TableName: common.TableName{ - Schema: "source", - Table: "dst", - IsPartition: true, - }, - }, - } - w.onDDL(ddl) - require.True(t, w.partitionTableAccessor.IsPartitionTable("target", "dst")) - - newDMLMessage := func(commitTs uint64) *codeccommon.DMLMessage { - return codeccommon.NewDMLMessage(1, "target", "dst", commitTs, common.RowTypeUpdate, nil) - } - - progress := w.progresses[0] - w.appendMessage2Group(newDMLMessage(200), progress) - w.appendMessage2Group(newDMLMessage(100), progress) - - resolved := progress.eventsGroup[1].ResolveInto(150, nil) - require.Len(t, resolved, 1) - require.Equal(t, uint64(100), resolved[0].GetCommitTs()) -} - func TestWriteMessageDefersDMLAssemblyUntilFlush(t *testing.T) { ctx := context.Background() ctrl := gomock.NewController(t)