DDLNotifier.processEvents Retries Infinitely
- Dominant language
- Go
- Stars
- 40.5k
- Forks
- 6.2k
- PR merge metrics
- PR metrics pending
Description
## Issue: DDLNotifier.processEvents Retries Infinitely
**Description**
The `DDLNotifier.processEvents` function in `pkg/ddl/notifier/subscribe.go` retries processing events indefinitely. If a handler consistently returns an error (other than `ErrNotReadyRetryLater`), the `processEvents` function will continuously log the error and retry without limit. This can lead to excessive logging, resource consumption, and potentially prevent the processing of other DDL events.
**Code Location**
`pkg/ddl/notifier/subscribe.go`: lines 158-194
**Steps to Reproduce**
1. Register a `SchemaChangeHandler` with the `DDLNotifier`.
2. Ensure the handler returns an error consistently (other than `ErrNotReadyRetryLater`).
3. Observe the logs and resource consumption as `processEvents` retries indefinitely.
**Expected Behavior**
The `processEvents` function should have a mechanism to limit the number of retries for a given event handler. After exceeding the retry limit, the event should be handled in a way that prevents infinite retries, such as logging the error and skipping the event for that handler.
**Proposed Solutions**
1. **Retry Counter:** Introduce a retry counter for each handler within the `processEvents` function. Increment the counter each time a handler returns an error. If the counter exceeds a predefined threshold, skip the handler for the current event and log an error.
```go
// filepath: pkg/ddl/notifier/subscribe.go
...existing code...
skipHandlers := make(map[HandlerID]struct{})
retryCounts := make(map[HandlerID]int) // Add retry counter
changes := make([]*SchemaChange, ProcessEventsBatchSize)
for {
count, err2 := result.Read(changes)
if err2 != nil {
return errors.Trace(err2)
}
if count == 0 {
break
}
for _, change := range changes[:count] {
for handlerID, handler := range n.handlers {
if _, ok := skipHandlers[handlerID]; ok {
continue
}
// Check retry count
const maxRetries = 3 // Define a maximum retry limit
if retryCounts[handlerID] >= maxRetries {
logutil.Logger(ctx).Error("Exceeded max retries for handler, skipping", zap.Stringer("handler", handlerID), zap.Int64("ddlJobID", change.ddlJobID), zap.Int64("subJobID", change.subJobID))
skipHandlers[handlerID] = struct{}{}
continue
}
if err3 := n.processEventForHandler(ctx, sess4Process, change, handlerID, handler); err3 != nil {
skipHandlers[handlerID] = struct{}{}
retryCounts[handlerID]++ // Increment retry counter
if !goerr.Is(err3, ErrNotReadyRetryLater) {
logutil.Logger(ctx).Error("Error processing change",
zap.Int64("ddlJobID", change.ddlJobID),
zap.Int64("subJobID", change.subJobID),
zap.Stringer("handler", handlerID),
zap.Error(err3))
}
continue
} else {
retryCounts[handlerID] = 0 // Reset retry counter on success
}
}
...existing code...
```
**Impact**
Without a retry limit, the `DDLNotifier` can become unstable and consume excessive resources when encountering persistent errors in event handlers. Implementing a retry limit or alternative error handling mechanism will improve the robustness and stability of the DDL notification system.
Contributor guide
Assessment
This issue has not been assessed yet.