Skip to content

consumer: sort fallback DML before flush (#5824) #5833

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

Open
ti-chi-bot wants to merge 1 commit into
base: release-8.5
Choose a base branch
from
Open

consumer: sort fallback DML before flush (#5824) #5833

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
45 changes: 45 additions & 0 deletions cmd/kafka-consumer/writer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -612,11 +612,26 @@ func (w *writer) appendRow2Group(dml *event.DMLEvent, progress *partitionProgres
table = dml.TableInfo.GetTableName()
commitTs = dml.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
}
<<<<<<< HEAD
// IMPORTANT: Kafka offsets are append-only, but CommitTs can go backwards after
// a TiCDC restart/retry (at-least-once replay). We must not drop such events
// solely based on a "seen" watermark (e.g. HighWatermark). The only safe
Expand Down Expand Up @@ -650,6 +665,36 @@ func (w *writer) appendRow2Group(dml *event.DMLEvent, progress *partitionProgres
zap.Uint64("appliedWatermark", group.AppliedWatermark),
zap.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
zap.Stringer("eventType", dml.RowTypes[0]))
=======
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, 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.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 {
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.String("schema", schema), zap.String("table", table), zap.Int64("tableID", tableID),
zap.Stringer("eventType", message.RowType))
return
}
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))
>>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824))
}

func openDB(ctx context.Context, dsn string) (*sql.DB, error) {
Expand Down
133 changes: 133 additions & 0 deletions cmd/kafka-consumer/writer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,7 @@ 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 TestAppendRow2Group_DoesNotDropCommitTsFallbackBeforeApplied(t *testing.T) {
// Scenario:
// 1) TiCDC writes DML messages to Kafka in commitTs order.
Expand All @@ -293,6 +294,99 @@ func TestAppendRow2Group_DoesNotDropCommitTsFallbackBeforeApplied(t *testing.T)
// The kafka-consumer must not drop these "fallback commitTs" events unless they have
// already been flushed to downstream (AppliedWatermark), otherwise the replay cannot
// heal the missing window.
=======
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) {
>>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824))
replicaCfg := config.GetDefaultReplicaConfig()
eventRouter, err := eventrouter.NewEventRouter(replicaCfg.Sink, "test-topic", false, false)
require.NoError(t, err)
Expand Down Expand Up @@ -340,3 +434,42 @@ func TestAppendRow2Group_DoesNotDropCommitTsFallbackBeforeApplied(t *testing.T)
resolved = group.ResolveInto(150, resolvedEvents)
require.Empty(t, resolved)
}

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
}
39 changes: 39 additions & 0 deletions cmd/pulsar-consumer/writer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -505,11 +505,25 @@ func (w *writer) appendRow2Group(dml *commonEvent.DMLEvent, progress *partitionP
table = dml.TableInfo.GetTableName()
commitTs = dml.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
}
<<<<<<< HEAD
if commitTs <= group.AppliedWatermark {
log.Warn("DML event replayed after applied, ignore it",
zap.Int64("tableID", tableID), zap.Int32("partition", group.Partition),
Expand All @@ -523,6 +537,21 @@ func (w *writer) appendRow2Group(dml *commonEvent.DMLEvent, progress *partitionP
if forceInsert {
log.Warn("DML event commit ts fallback, append with forceInsert",
zap.Int32("partition", group.Partition),
=======
group.AppendMessage(message)
if commitTs < progress.watermark {
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.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 {
log.Debug("DML event append to the group",
>>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824))
zap.Uint64("commitTs", commitTs), zap.Uint64("highWatermark", group.HighWatermark),
zap.Uint64("appliedWatermark", group.AppliedWatermark),
zap.Uint64("partitionWatermark", progress.watermark),
Expand All @@ -532,11 +561,21 @@ func (w *writer) appendRow2Group(dml *commonEvent.DMLEvent, progress *partitionP
group.Append(dml, true)
return
}
<<<<<<< HEAD
group.Append(dml, false)
log.Info("DML event append to the group",
zap.Int32("partition", group.Partition),
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]))
=======
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))
>>>>>>> af33cc193 (consumer: sort fallback DML before flush (#5824))
}
Loading
Loading