Skip to content

consumer: use dml message instead of dml event (#5590) #5778

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ti-chi-bot merged 4 commits into from
Aug 3, 2026
Merged

consumer: use dml message instead of dml event (#5590) #5778

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7426157
This is an automated cherry-pick of #5590
wk989898 Jul 21, 2026
2bcf842
update
wk989898 Aug 3, 2026
3b5a2db
update
wk989898 Aug 3, 2026
fa14dbf
Merge branch 'release-8.5' into cherry-pick-5590-to-release-8.5
wk989898 Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
188 changes: 124 additions & 64 deletions cmd/kafka-consumer/writer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,6 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error {
var (
done = make(chan struct{}, 1)

total int
flushed atomic.Int64
)

Expand All @@ -171,24 +170,16 @@ func (w *writer) flushDDLEvent(ctx context.Context, ddl *event.DDLEvent) error {
if !ok {
continue
}
before := len(resolvedEvents)
resolvedEvents = g.ResolveInto(commitTs, resolvedEvents)
resolvedCount := len(resolvedEvents) - before
if resolvedCount == 0 {
continue
messages := g.ResolveInto(commitTs, nil)
events := make([]*event.DMLEvent, 0, len(messages))
for _, message := range messages {
events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent())
}

resolvedGroups = append(resolvedGroups, struct {
group *util.EventsGroup
maxCommitTs uint64
}{
group: g,
maxCommitTs: resolvedEvents[len(resolvedEvents)-1].GetCommitTs(),
})
total += resolvedCount
resolvedEvents = append(resolvedEvents, events...)
}
}

total := len(resolvedEvents)
if total == 0 {
return w.mysqlSink.WriteBlockEvent(ddl)
}
Expand Down Expand Up @@ -289,7 +280,6 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error {
var (
done = make(chan struct{}, 1)

total int
flushed atomic.Int64
)

Expand All @@ -303,23 +293,15 @@ func (w *writer) flushDMLEventsByWatermark(ctx context.Context) error {
}, 0)
for _, p := range w.progresses {
for _, group := range p.eventsGroup {
before := len(resolvedEvents)
resolvedEvents = group.ResolveInto(watermark, resolvedEvents)
resolvedCount := len(resolvedEvents) - before
if resolvedCount == 0 {
continue
messages := group.ResolveInto(watermark, nil)
events := make([]*event.DMLEvent, 0, len(messages))
for _, message := range messages {
events = util.AppendOrMergeDMLEvent(events, message.ToDMLEvent())
}

resolvedGroups = append(resolvedGroups, struct {
group *util.EventsGroup
maxCommitTs uint64
}{
group: group,
maxCommitTs: resolvedEvents[len(resolvedEvents)-1].GetCommitTs(),
})
total += resolvedCount
resolvedEvents = append(resolvedEvents, events...)
}
}
total := len(resolvedEvents)
if total == 0 {
return nil
}
Expand Down Expand Up @@ -391,12 +373,12 @@ func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool
ddl := progress.decoder.NextDDLEvent()

if dec, ok := progress.decoder.(*simple.Decoder); ok {
cachedEvents := dec.GetCachedEvents()
for _, row := range cachedEvents {
cachedMessages := dec.GetCachedMessages()
for _, dmlMessage := range cachedMessages {
log.Info("simple protocol cached event resolved, append to the group",
zap.Int64("tableID", row.GetTableID()), zap.Uint64("commitTs", row.CommitTs),
zap.Int64("tableID", dmlMessage.TableID), zap.Uint64("commitTs", dmlMessage.GetCommitTs()),
zap.Int32("partition", partition), zap.Any("offset", offset))
w.appendRow2Group(row, progress, offset)
w.appendMessage2Group(dmlMessage, progress, offset)
}
}

Expand All @@ -420,25 +402,33 @@ func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool
needFlush = true
case common.MessageTypeRow:
var counter int
row := progress.decoder.NextDMLEvent()
if row == nil {
dmlMessage := progress.decoder.NextDMLMessage()
if dmlMessage == nil {
if w.protocol != config.ProtocolSimple {
log.Panic("DML event is nil, it's not expected",
log.Panic("DML message is nil, it's not expected",
zap.Int32("partition", partition), zap.Any("offset", offset))
}
log.Debug("DML event is nil, it's cached", zap.Int32("partition", partition), zap.Any("offset", offset))
log.Debug("DML message is nil, it's cached", zap.Int32("partition", partition), zap.Any("offset", offset))
break
}

w.appendRow2Group(row, progress, offset)
w.appendMessage2Group(dmlMessage, progress, offset)
counter++
for {
_, hasNext = progress.decoder.HasNext()
if !hasNext {
break
}
row = progress.decoder.NextDMLEvent()
w.appendRow2Group(row, progress, offset)
dmlMessage = progress.decoder.NextDMLMessage()
if dmlMessage == nil {
if w.protocol != config.ProtocolSimple {
log.Panic("DML message is nil, it's not expected",
zap.Int32("partition", partition), zap.Any("offset", offset))
}
log.Debug("DML message is nil, it's cached", zap.Int32("partition", partition), zap.Any("offset", offset))
break
}
w.appendMessage2Group(dmlMessage, progress, offset)
counter++
}
// If the message containing only one event exceeds the length limit, CDC will allow it and issue a warning.
Expand Down Expand Up @@ -541,21 +531,25 @@ func (w *writer) onDDL(ddl *event.DDLEvent) {
return
}
switch w.protocol {
case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, config.ProtocolSimple, config.ProtocolDebezium:
case config.ProtocolCanalJSON, config.ProtocolOpen, config.ProtocolAvro, config.ProtocolSimple,
config.ProtocolDebezium:
default:
return
}
// TODO: support more corner cases
// e.g. create partition table + drop table(rename table) + create normal table: the partitionTableAccessor should drop the table when the table become normal.
switch model.ActionType(ddl.Type) {
case model.ActionCreateTable:
if w.markPartitionTableFromDDL(ddl) {
return
}
stmt, err := parser.New().ParseOneStmt(ddl.Query, "", "")
if err != nil {
log.Panic("parse ddl query failed", zap.String("query", ddl.Query), zap.Error(err))
}
if v, ok := stmt.(*ast.CreateTableStmt); ok {
if v.Partition != nil {
w.partitionTableAccessor.Add(ddl.GetSchemaName(), ddl.GetTableName())
w.addPartitionTable(ddl.GetSchemaName(), ddl.GetTableName())
return
}
if v.ReferTable != nil {
Expand All @@ -564,17 +558,36 @@ func (w *writer) onDDL(ddl *event.DDLEvent) {
referSchema = ddl.GetSchemaName()
}
if w.partitionTableAccessor.IsPartitionTable(referSchema, v.ReferTable.Name.O) {
w.partitionTableAccessor.Add(ddl.GetSchemaName(), ddl.GetTableName())
w.addPartitionTable(ddl.GetSchemaName(), ddl.GetTableName())
}
}
}
case model.ActionRenameTable:
if w.partitionTableAccessor.IsPartitionTable(ddl.ExtraSchemaName, ddl.ExtraTableName) {
w.partitionTableAccessor.Add(ddl.GetSchemaName(), ddl.GetTableName())
w.addPartitionTable(ddl.GetSchemaName(), ddl.GetTableName())
}
w.markPartitionTableFromDDL(ddl)
}
}

func (w *writer) markPartitionTableFromDDL(ddl *event.DDLEvent) bool {
if ddl.TableInfo == nil || !ddl.TableInfo.IsPartitionTable() {
return false
}

w.addPartitionTable(ddl.GetSchemaName(), ddl.GetTableName())
w.addPartitionTable(ddl.TableInfo.GetSchemaName(), ddl.TableInfo.GetTableName())
w.addPartitionTable(ddl.TableInfo.GetTargetSchemaName(), ddl.TableInfo.GetTargetTableName())
return true
}

func (w *writer) addPartitionTable(schema, table string) {
if schema == "" || table == "" {
return
}
w.partitionTableAccessor.Add(schema, table)
}

func (w *writer) checkPartition(row *event.DMLEvent, partition int32, offset kafka.Offset) {
var (
partitioner = w.eventRouter.GetPartitionGenerator(row.TableInfo.GetSchemaName(), row.TableInfo.GetTableName())
Expand Down Expand Up @@ -602,15 +615,22 @@ func (w *writer) checkPartition(row *event.DMLEvent, partition int32, offset kaf
}
}

func (w *writer) appendRow2Group(dml *event.DMLEvent, progress *partitionProgress, offset kafka.Offset) {
w.checkPartition(dml, progress.partition, offset)
func (w *writer) messageWithPartitionCheck(message *common.DMLMessage, partition int32, offset kafka.Offset) *common.DMLMessage {
return common.NewDMLMessage(message.TableID, message.Schema, message.Table, message.GetCommitTs(), message.RowType, func() *event.DMLEvent {
row := message.ToDMLEvent()
w.checkPartition(row, partition, offset)
return row
})
}

func (w *writer) appendMessage2Group(message *common.DMLMessage, progress *partitionProgress, offset kafka.Offset) {
// if the kafka cluster is normal, this should not hit.
// else if the cluster is abnormal, the consumer may consume old message, then cause the watermark fallback.
var (
tableID = dml.GetTableID()
schema = dml.TableInfo.GetSchemaName()
table = dml.TableInfo.GetTableName()
commitTs = dml.GetCommitTs()
tableID = message.TableID
schema = message.Schema
table = message.Table
commitTs = message.GetCommitTs()
)
group := progress.eventsGroup[tableID]
if group == nil {
Expand All @@ -630,26 +650,66 @@ func (w *writer) appendRow2Group(dml *event.DMLEvent, progress *partitionProgres
zap.String("schema", schema), zap.String("table", table), zap.Any("protocol", w.protocol))
return
}
forceInsert := commitTs < group.HighWatermark || commitTs < progress.watermark || w.enableTableAcrossNodes
if forceInsert {
log.Warn("DML event commit ts fallback, append with forceInsert",
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),
zap.Uint64("appliedWatermark", group.AppliedWatermark),
zap.Uint64("partitionWatermark", progress.watermark), zap.Any("watermarkOffset", progress.watermarkOffset),
zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
zap.Stringer("eventType", dml.RowTypes[0]), zap.Any("protocol", w.protocol),
zap.Bool("IsPartition", dml.TableInfo.TableName.IsPartition))
group.Append(dml, true)
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
}
group.Append(dml, false)
log.Info("DML event append to the group",
zap.Int32("partition", group.Partition), zap.Any("offset", offset),
zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
zap.Uint64("appliedWatermark", group.AppliedWatermark),
zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
zap.Stringer("eventType", dml.RowTypes[0]))
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:
// 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))
}
}

func openDB(ctx context.Context, dsn string) (*sql.DB, error) {
Expand Down
Loading
Loading