Skip to content
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

[DNM] consumer(ticdc): remove order message #11812

Open
wants to merge 23 commits into
base: master
Choose a base branch
from
+184 −107
Open

[DNM] consumer(ticdc): remove order message #11812

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
59c45c1
remove order events
3AceShowHand Nov 27, 2024
049849a
check offset on the kafka sink
3AceShowHand Nov 27, 2024
70400f9
add more logs to consumer to panic
3AceShowHand Nov 27, 2024
2c3dcc4
add more logs to consumer to panic
3AceShowHand Nov 27, 2024
8c37f90
fix the worker
3AceShowHand Nov 27, 2024
4a62446
fix the worker
3AceShowHand Nov 28, 2024
50ff9e6
add more logs
3AceShowHand Nov 28, 2024
7426af0
add more logs
3AceShowHand Nov 28, 2024
185e17b
fix check
3AceShowHand Nov 28, 2024
a1ed897
add table id to consumer log
3AceShowHand Nov 28, 2024
9011000
add partiton to log
3AceShowHand Nov 28, 2024
cf55d34
add more logs
3AceShowHand Nov 29, 2024
b268de7
remove resolveTs
wk989898 Nov 29, 2024
54fa84c
fix commit ts
wk989898 Nov 29, 2024
322c8ec
add more
3AceShowHand Nov 29, 2024
98517af
adjust writer
3AceShowHand Dec 2, 2024
8ac2c78
also flush when meet row
3AceShowHand Dec 2, 2024
c0d25d9
add more logs
wk989898 Dec 2, 2024
1b7c2b2
also flush if exceed 2 minutes
3AceShowHand Dec 2, 2024
75c1fea
Merge branch 'kafka-sarama-no-retry' of https://github.com/3AceShowHa…
3AceShowHand Dec 2, 2024
0f78ecc
also print es
3AceShowHand Dec 2, 2024
4c5b8c5
flush event immediately
3AceShowHand Dec 2, 2024
49cfd06
flush immediately
3AceShowHand Dec 2, 2024
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
Prev Previous commit
Next Next commit
add more logs
  • Loading branch information
@3AceShowHand
3AceShowHand committed Nov 29, 2024
commit cf55d342f20701ce9c032a81a516db6fc0e88aa5
2 changes: 2 additions & 0 deletions cdc/sink/dmlsink/mq/worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -307,6 +307,7 @@ func (w *worker) sendMessages(ctx context.Context) error {
log.Warn("Ts not monotonically increasing",
zap.String("namespace", w.changeFeedID.Namespace),
zap.String("changefeed", w.changeFeedID.ID),
zap.Int32("partition", future.Key.Partition),
zap.Int64("tableID", event.Event.GetTableID()),
zap.Uint64("commitTs", event.Event.CommitTs),
zap.Int("length", len(future.Events)))
Expand All @@ -315,6 +316,7 @@ func (w *worker) sendMessages(ctx context.Context) error {
log.Panic("Ts is not monotonically increasing",
zap.String("namespace", w.changeFeedID.Namespace),
zap.String("changefeed", w.changeFeedID.ID),
zap.Int32("partition", future.Key.Partition),
zap.Int64("tableID", previousMessage.TableID),
zap.Uint64("previousTs", previousMessage.Ts),
zap.Int64("currentTableID", message.TableID),
Expand Down
2 changes: 1 addition & 1 deletion cmd/kafka-consumer/consumer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,7 +136,7 @@ func (c *consumer) Consume(ctx context.Context) {
log.Error("commit message failed, just continue", zap.Error(err))
continue
}
log.Info("commit message success",
log.Debug("commit message success",
zap.String("topic", topicPartition[0].String()), zap.Int32("partition", topicPartition[0].Partition),
zap.Any("offset", topicPartition[0].Offset))
}
Expand Down
19 changes: 9 additions & 10 deletions cmd/kafka-consumer/writer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -387,15 +387,16 @@ func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool

prev, ok := progress.previousMap[tableID]
if ok {
if prev.commitTs > row.CommitTs {
if prev.offset < message.TopicPartition.Offset {
if prev.commitTs > row.CommitTs && prev.offset < message.TopicPartition.Offset {
watermark := atomic.LoadUint64(&progress.watermark)
if row.CommitTs < atomic.LoadUint64(&progress.watermark) {
log.Panic("row changed event commitTs fallback",
zap.Int64("tableID", tableID), zap.Int32("partition", partition),
zap.Int64("tableID", tableID), zap.Int32("partition", partition), zap.Uint64("watermark", watermark),
zap.Uint64("previous", prev.commitTs), zap.Uint64("commitTs", row.CommitTs),
zap.Any("previousOffset", prev.offset), zap.Any("offset", message.TopicPartition.Offset))
}
log.Warn("row changed event commitTs fallback, ignore it",
zap.Int64("tableID", tableID), zap.Int32("partition", partition),
zap.Int64("tableID", tableID), zap.Int32("partition", partition), zap.Uint64("watermark", watermark),
zap.Uint64("previous", prev.commitTs), zap.Uint64("commitTs", row.CommitTs),
zap.Any("previousOffset", prev.offset), zap.Any("offset", message.TopicPartition.Offset))
}
Expand All @@ -411,14 +412,12 @@ func (w *writer) WriteMessage(ctx context.Context, message *kafka.Message) bool
eventGroup[tableID] = group
}
group.Append(row)
log.Debug("DML event received",
log.Info("DML event received",
zap.String("schema", row.TableInfo.GetSchemaName()),
zap.String("table", row.TableInfo.GetTableName()),
zap.Int32("partition", partition),
zap.Any("offset", message.TopicPartition.Offset),
zap.Uint64("commitTs", row.CommitTs),
zap.Int64("physicalTableID", row.PhysicalTableID),
zap.Int64("tableID", tableID),
zap.String("schema", row.TableInfo.GetSchemaName()),
zap.String("table", row.TableInfo.GetTableName()))
zap.Uint64("commitTs", row.CommitTs))
case model.MessageTypeResolved:
ts, err := decoder.NextResolvedEvent()
if err != nil {
Expand Down
4 changes: 3 additions & 1 deletion pkg/sink/kafka/factory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,10 +248,12 @@ func (p *saramaAsyncProducer) AsyncRunCallback(
commitTs := string(ack.Headers[1].Value)
previous := memo[ack.Partition]
log.Info("async producer receive ack",
zap.Int32("partition", ack.Partition),
zap.Int64("oldOffset", previous.offset), zap.String("oldCommitTs", previous.commitTs), zap.String("oldTableID", previous.tableID),
zap.Int64("newOffset", ack.Offset), zap.String("commitTs", commitTs), zap.String("tableID", tableID))
if ack.Offset < previous.offset {
log.Warn("kafka async producer receive an out-of-order message",
zap.Int32("partition", ack.Partition),
zap.Int64("oldOffset", previous.offset), zap.String("oldCommitTs", previous.commitTs), zap.String("oldTableID", previous.tableID),
zap.Int64("newOffset", ack.Offset), zap.String("commitTs", commitTs), zap.String("tableID", tableID))
}
Expand Down Expand Up @@ -296,7 +298,7 @@ func (p *saramaAsyncProducer) AsyncSend(ctx context.Context, topic string, parti
case p.producer.Input() <- msg:
}
log.Info("async producer send message",
zap.Int64("tableID", message.TableID), zap.Uint64("commitTs", message.Ts))
zap.Int64("tableID", message.TableID), zap.Int32("partition", partition), zap.Uint64("commitTs", message.Ts))

return nil
}
Loading