Fix bugs in merge operation #1436
Conversation
There was a problem hiding this comment.
Summary of Changes
Hello @hongyunyan, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request addresses a specific issue that can occur when merging dispatchers in the system, particularly when DDL events are involved. The problem arises because if dispatchers are merged before a DDL event is fully processed and flushed downstream, the newly merged dispatcher might start processing subsequent DMLs using the old schema, leading to errors. The solution introduces a mechanism to ensure that during the merge process, the dispatchers being closed wait for any pending DDL events to be flushed, guaranteeing that the merged dispatcher starts from a consistent state with the correct schema.
Highlights
- Fix Corner Case in Dispatcher Merging: Modified the
TryClosefunction in the dispatcher to accept a boolean flag (waitDDL) to control whether it should wait for DDL events to be flushed before indicating it's ready to close. This addresses a corner case during dispatcher merging where not waiting for DDLs could lead to the merged dispatcher processing DMLs with an outdated schema. - Conditional DDL Wait: The new
waitDDLflag is set totruespecifically whenTryCloseis called during the dispatcher merge process (DoMergefunction). In all other scenarios (general removal, closing all dispatchers), the flag is set tofalse, maintaining the previous behavior.
Using Gemini Code Assist
The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.
Invoking Gemini
You can request assistance from Gemini at any point in your pull request via creating an issue comment (i.e. comment on the pull request page) using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands.
| Feature | Command | Description |
|---|---|---|
| Code Review | /gemini review |
Performs a code review for the current pull request in its current state. |
| Pull Request Summary | /gemini summary |
Provides a summary of the current pull request in its current state. |
| Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in issue comments and review comments. |
| Help | /gemini help |
Displays a list of available commands. |
Customization
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.
Limitations & Feedback
Gemini Code Assist is currently in preview and may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments to provide feedback.
You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.
Footnotes
-
Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configureGemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩
There was a problem hiding this comment.
Code Review
The pull request introduces a new parameter waitDDL to the TryClose function of the Dispatcher to handle a corner case when merging dispatchers near some DDL events. The changes also include modifications to the EventDispatcherManager to use this new parameter and updates to the tests.
| @@ -383,7 +383,17 @@ func (d *Dispatcher) isFirstEvent(event commonEvent.Event) bool { | |||
| // TryClose should be called before Remove(), because the dispatcher may still wait the dispatcher status from maintainer. | |||
| // TryClose will return the watermark of current dispatcher, and return true when the dispatcher finished sending events to sink. | |||
| // EventDispatcherManager will clean the dispatcher info after Remove() is called. | |||
| func (d *Dispatcher) TryClose() (w heartbeatpb.Watermark, ok bool) { | |||
| // | |||
| // If waitDDL is true, dispatcher can't wait ddl event to be flushed or pass, which means, dispatcher's blockEventStatus should be nil | |||
There was a problem hiding this comment.
It would be helpful to add a comment explaining why blockEventStatus should be nil when waitDDL is true.
| // If waitDDL is true, dispatcher can't wait ddl event to be flushed or pass, which means, dispatcher's blockEventStatus should be nil | |
| // If waitDDL is true, dispatcher can't wait ddl event to be flushed or pass, which means, dispatcher's blockEventStatus should be nil | |
| // This is because when merging dispatchers, we need to ensure the final startTs of the merged dispatcher | |
| // is not less than a flushed ddl event in the dispatchers to be merged. |
10000
| @@ -821,7 +821,7 @@ func (e *EventDispatcherManager) DoMerge(t *MergeCheckTask) { | |||
| appcontext.GetService[*eventcollector.EventCollector](appcontext.EventCollector).RemoveDispatcher(dispatcher) | |||
| } | |||
|
|
|||
| watermark, ok := dispatcher.TryClose() | |||
| watermark, ok := dispatcher.TryClose(true) | |||
There was a problem hiding this comment.
Consider adding a log message here to indicate that the dispatcher is being closed with waitDDL = true.
log.Info("closing dispatcher with waitDDL", zap.Stringer("dispatcherID", id))
watermark, ok := dispatcher.TryClose(true)|
/test pull-cdc-mysql-integration-heavy |
|
@hongyunyan: The following test failed, say
Full PR test history. Your PR dashboard. Instructions 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. |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: asddongmen The full list of commands accepted by this bot can be found here. The pull request process is described here
Needs approval from an approver in each of these files:
Approvers can indicate their approval by writing |
[LGTM Timeline notifier]Timeline:
|
What problem does this PR solve?
Issue Number: close #1434
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