Private/issues 543 canal cache memory#1156
Conversation
docs: update local PR description with full-suite validation
docs: align local PR notes with strict test policy
![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 introduces an LRU-based table metadata cache with a configurable capacity limit to prevent unbounded memory growth, and implements automatic recovery mechanisms when replication encounters an 'impossible position' error. The review feedback highlights several improvement opportunities: preserving the original error during fallback failures in startSyncer() to avoid misleading logs, simplifying redundant logic in isImpossibleBinlogPositionError(), avoiding unnecessary write-lock contention in GetTable() when the cache is disabled, and removing an unused parameter from setTableCacheLocked() along with updating its call sites.
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.
| s, err := c.syncer.StartSync(pos) | ||
| if err != nil { | ||
| if isImpossibleBinlogPositionError(err) { | ||
| fallbackPos, fallbackErr := c.GetMasterPos() | ||
| if fallbackErr != nil { | ||
| return nil, errors.Errorf("start sync replication at binlog %v error %v; fallback to master position failed: %v", pos, err, fallbackErr) | ||
| } | ||
|
|
||
| c.cfg.Logger.Warn("requested binlog position is impossible, falling back to current master position", | ||
| slog.Any("requested_pos", pos), | ||
| slog.Any("fallback_pos", fallbackPos), | ||
| slog.Any("error", err), | ||
| ) | ||
|
|
||
| s, err = c.syncer.StartSync(fallbackPos) | ||
| if err == nil { | ||
| c.master.Update(fallbackPos) | ||
| c.cfg.Logger.Info("start sync binlog at fallback binlog file", slog.Any("pos", fallbackPos)) | ||
| return s, nil | ||
| } | ||
| c.cfg.Logger.Warn("fallback start sync failed, returning original startup error", | ||
| slog.Any("requested_pos", pos), | ||
| slog.Any("fallback_pos", fallbackPos), | ||
| slog.Any("fallback_error", err), | ||
| ) | ||
| } | ||
| return nil, errors.Errorf("start sync replication at binlog %v error %v", pos, err) | ||
| } |
There was a problem hiding this comment.
In startSyncer(), if the fallback to the master position fails, the original error err is overwritten by the fallback error. As a result, the function returns the fallback error but associates it with the original position pos in the error message, which is misleading and contradicts the logged warning "fallback start sync failed, returning original startup error". We should preserve the original error and return it if the fallback fails.
s, err := c.syncer.StartSync(pos)
if err != nil {
if isImpossibleBinlogPositionError(err) {
fallbackPos, fallbackErr := c.GetMasterPos()
if fallbackErr != nil {
return nil, errors.Errorf("start sync replication at binlog %v error %v; fallback to master position failed: %v", pos, err, fallbackErr)
}
c.cfg.Logger.Warn("requested binlog position is impossible, falling back to current master position",
slog.Any("requested_pos", pos),
slog.Any("fallback_pos", fallbackPos),
slog.Any("error", err),
)
fallbackStreamer, fallbackSyncErr := c.syncer.StartSync(fallbackPos)
if fallbackSyncErr == nil {
c.master.Update(fallbackPos)
c.cfg.Logger.Info("start sync binlog at fallback binlog file", slog.Any("pos", fallbackPos))
return fallbackStreamer, nil
}
c.cfg.Logger.Warn("fallback start sync failed, returning original startup error",
slog.Any("requested_pos", pos),
slog.Any("fallback_pos", fallbackPos),
slog.Any("fallback_error", fallbackSyncErr),
)
}
return nil, errors.Errorf("start sync replication at binlog %v error %v", pos, err)
}| func isImpossibleBinlogPositionError(err error) bool { | ||
| if err == nil { | ||
| return false | ||
| } | ||
|
|
||
| if mysql.ErrorCode(err.Error()) == mysql.ER_MASTER_FATAL_ERROR_READING_BINLOG { | ||
| errMsg := strings.ToLower(err.Error()) | ||
| return strings.Contains(errMsg, "impossible position") | ||
| } | ||
|
|
||
| errMsg := strings.ToLower(err.Error()) | ||
| return strings.Contains(errMsg, "impossible position") | ||
| } |
There was a problem hiding this comment.
The isImpossibleBinlogPositionError function contains redundant code. The if block checking mysql.ErrorCode performs the exact same string containment check as the fallback code outside the block. We can simplify this function to improve readability and maintainability.
func isImpossibleBinlogPositionError(err error) bool {
if err == nil {
return false
}
errMsg := strings.ToLower(err.Error())
return strings.Contains(errMsg, "impossible position")
}| if ok { | ||
| c.tableLock.Lock() | ||
| c.touchTableCacheLocked(key) | ||
| c.tableLock.Unlock() | ||
| return t, nil | ||
| } |
There was a problem hiding this comment.
When the table cache is unbounded (tableCacheCapacity <= 0), calling touchTableCacheLocked still acquires the write lock c.tableLock.Lock(), which can cause unnecessary lock contention on concurrent reads. We can safely check c.tableCacheCapacity > 0 before acquiring the write lock to avoid this overhead.
if ok {
if c.tableCacheCapacity > 0 {
c.tableLock.Lock()
c.touchTableCacheLocked(key)
c.tableLock.Unlock()
}
return t, nil
}| func (c *Canal) setTableCacheLocked(key string, table *schema.Table) { | ||
| _ = table | ||
| if c.tableCacheCapacity <= 0 || c.tableCacheOrder == nil || c.tableCacheNodes == nil { | ||
| return | ||
| } |
There was a problem hiding this comment.
The table parameter in setTableCacheLocked is unused, as indicated by the _ = table workaround. We can remove this parameter to clean up the function signature.
func (c *Canal) setTableCacheLocked(key string) {
if c.tableCacheCapacity <= 0 || c.tableCacheOrder == nil || c.tableCacheNodes == nil {
return
}| c.tableLock.Lock() | ||
| c.tables[key] = ta | ||
| c.setTableCacheLocked(key, ta) | ||
| c.tableLock.Unlock() |
There was a problem hiding this comment.
Update the call to setTableCacheLocked to match the simplified signature without the unused table parameter.
| c.tableLock.Lock() | |
| c.tables[key] = ta | |
| c.setTableCacheLocked(key, ta) | |
| c.tableLock.Unlock() | |
| c.tableLock.Lock() | |
| c.tables[key] = ta | |
| c.setTableCacheLocked(key) | |
| c.tableLock.Unlock() |
| c.tableLock.Lock() | ||
| c.tables[key] = t | ||
| c.setTableCacheLocked(key, t) | ||
| if c.cfg.DiscardNoMetaRowEvent { |
There was a problem hiding this comment.
Update the call to setTableCacheLocked to match the simplified signature without the unused table parameter.
| c.tableLock.Lock() | |
| c.tables[key] = t | |
| c.setTableCacheLocked(key, t) | |
| if c.cfg.DiscardNoMetaRowEvent { | |
| c.tableLock.Lock() | |
| c.tables[key] = t | |
| c.setTableCacheLocked(key) | |
| if c.cfg.DiscardNoMetaRowEvent { |
| c.tableLock.Lock() | ||
| c.tables[key] = schema | ||
| c.setTableCacheLocked(key, schema) | ||
| if c.cfg.DiscardNoMetaRowEvent { |
There was a problem hiding this comment.
Update the call to setTableCacheLocked to match the simplified signature without the unused table parameter.
| c.tableLock.Lock() | |
| c.tables[key] = schema | |
| c.setTableCacheLocked(key, schema) | |
| if c.cfg.DiscardNoMetaRowEvent { | |
| c.tableLock.Lock() | |
| c.tables[key] = schema | |
| c.setTableCacheLocked(key) | |
| if c.cfg.DiscardNoMetaRowEvent { |
1 participant
PR Description: canal
RunFromimpossible binlog position handling (#642)Summary
This PR addresses canal startup failures when
RunFromis initialized with a binlog file/offset pair that is no longer valid on the upstream server.Typical error:
Client requested master to start replication from impossible position; the first event 'binlog.024078' at 1247892405, the last event read from 'binlog.024078' at 4, the last byte read from 'binlog.024078' at 4.The goal is to make startup behavior deterministic and recoverable for this class of invalid checkpoint.
Background / user impact
In production, teams often run multiple consumers against the same binlog stream for parallel processing.
When one consumer restarts with an outdated or inconsistent position, canal currently fails fast at startup and cannot continue syncing.
Common triggers:
Root cause
RunFromassumes the provided start position is valid for the current upstream binlog state.When MySQL rejects the requested position as impossible, startup aborts instead of entering a controlled fallback path.
Proposed changes
RunFromposition is valid.Behavior after fix
RunFrominputs: no behavior change.Compatibility
RunFrom.Validation plan
RunFrompositions still behave identically.Risk and mitigation
Risk: selecting an incorrect fallback point in edge cases.
Mitigation:
Rollout notes
Issue linkage