consumer: use dml message instead of dml event#5590
Conversation
ti-chi-bot
Bot
added
do-not-merge/work-in-progress
|
Skipping CI for Draft Pull Request. |
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ti-chi-bot
Bot
added
the
size/XXL
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the consumer and codec components to use a deferred DMLMessage abstraction instead of assembling DMLEvent objects immediately, which optimizes memory usage and performance. Additionally, the Canal-JSON decoder is updated to track DDL commit timestamps for more accurate table schema caching across column changes. The code review feedback suggests several defensive programming enhancements to prevent potential nil pointer dereferences and nil map panics, specifically by checking for nil messages in consumer.go and event_group.go, and ensuring the ddlCommitTs map is properly initialized and checked in canal_json_decoder.go.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| row := decoder.NextDMLMessage().ToDMLEvent() | ||
| row.PhysicalTableID = tableID | ||
| c.appendRow2Group(row, fileIdx.EnableTableAcrossNodes) | ||
| c.appendMessage2Group(common.NewDMLMessageFromEvent(row), fileIdx.EnableTableAcrossNodes) |
There was a problem hiding this comment.
For defensive programming, we should check if decoder.NextDMLMessage() is nil before calling .ToDMLEvent() to avoid any potential nil pointer dereference panic.
dmlMsg := decoder.NextDMLMessage()
if dmlMsg == nil {
log.Panic("DML message is nil, it's not expected")
}
row := dmlMsg.ToDMLEvent()
row.PhysicalTableID = tableID
c.appendMessage2Group(common.NewDMLMessageFromEvent(row), fileIdx.EnableTableAcrossNodes)References
- Defensive programming: always check for nil pointers before dereferencing or calling methods on objects that can be nil.
| func (d *decoder) addDDLCommitTs(schema, table string, commitTs uint64) { | ||
| if schema == "" || table == "" || commitTs == 0 { | ||
| return | ||
| } |
There was a problem hiding this comment.
If d.ddlCommitTs is not initialized (e.g., if the decoder is created manually in tests or other packages without calling NewDecoder), calling d.ddlCommitTs[key] will panic. We should defensively initialize it if it is nil.
func (d *decoder) addDDLCommitTs(schema, table string, commitTs uint64) {
if schema == "" || table == "" || commitTs == 0 {
return
}
if d.ddlCommitTs == nil {
d.ddlCommitTs = make(map[tableNameKey][]uint64)
}
key := tableNameKey{schema: schema, table: table}References
- Defensive programming: ensure maps are initialized before writing to them to avoid nil map panics.
| func (d *decoder) getDDLCommitTs(schema, table string, commitTs uint64) uint64 { | ||
| if commitTs == 0 { | ||
| return 0 | ||
| } | ||
|
|
||
| commitTsList := d.ddlCommitTs[tableNameKey{schema: schema, table: table}] |
There was a problem hiding this comment.
If d.ddlCommitTs is nil, accessing it will panic. We should defensively check if d.ddlCommitTs is nil before querying it.
| func (d *decoder) getDDLCommitTs(schema, table string, commitTs uint64) uint64 { | |
| if commitTs == 0 { | |
| return 0 | |
| } | |
| commitTsList := d.ddlCommitTs[tableNameKey{schema: schema, table: table}] | |
| func (d *decoder) getDDLCommitTs(schema, table string, commitTs uint64) uint64 { | |
| if commitTs == 0 || d.ddlCommitTs == nil { | |
| return 0 | |
| } | |
| commitTsList := d.ddlCommitTs[tableNameKey{schema: schema, table: table}] |
References
- Defensive programming: check for nil maps before reading from them to avoid nil pointer panics.
| func AppendOrMergeDMLEvent(events []*commonEvent.DMLEvent, row *commonEvent.DMLEvent) []*commonEvent.DMLEvent { | ||
| var lastDMLEvent *commonEvent.DMLEvent | ||
| if len(events) > 0 { | ||
| lastDMLEvent = events[len(events)-1] | ||
| } | ||
|
|
||
| if lastDMLEvent == nil || lastDMLEvent.GetCommitTs() < row.GetCommitTs() { |
There was a problem hiding this comment.
If row is nil, calling row.GetCommitTs() will panic with a nil pointer dereference. We should defensively check if row is nil at the beginning of AppendOrMergeDMLEvent.
func AppendOrMergeDMLEvent(events []*commonEvent.DMLEvent, row *commonEvent.DMLEvent) []*commonEvent.DMLEvent {
if row == nil {
return events
}
var lastDMLEvent *commonEvent.DMLEvent
if len(events) > 0 {
lastDMLEvent = events[len(events)-1]
}
if lastDMLEvent == nil || lastDMLEvent.GetCommitTs() < row.GetCommitTs() {References
- Defensive programming: check for nil arguments before calling methods on them to avoid nil pointer panics.
|
/test kafka |
|
/test kafka |
|
@wk989898: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
1 participant
What problem does this PR solve?
Issue Number: close #5587
What is changed and how it works?
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note