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

kafka(ticdc): sarama do not retry if produce message failed to prevent out of order (#11870) #11960

Open
wants to merge 3 commits into
base: release-7.5
Choose a base branch
from
+9 −9
Open

kafka(ticdc): sarama do not retry if produce message failed to prevent out of order (#11870) #11960

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6f0dd2d
set the sarama producer retry to 0
3AceShowHand Dec 12, 2024
050f76d
enable the sarama log
3AceShowHand Dec 26, 2024
3569b08
fix make fmt
3AceShowHand Dec 31, 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
11 changes: 4 additions & 7 deletions pkg/logutil/log.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,14 +223,11 @@ func initMySQLLogger() error {

// initSaramaLogger hacks logger used in sarama lib
func initSaramaLogger(level zapcore.Level) error {
// only available less than info level
if !zapcore.InfoLevel.Enabled(level) {
logger, err := zap.NewStdLogAt(log.L().With(zap.String("component", "sarama")), level)
if err != nil {
return errors.Trace(err)
}
sarama.Logger = logger
logger, err := zap.NewStdLogAt(log.L().With(zap.String("component", "sarama")), level)
if err != nil {
return errors.Trace(err)
}
sarama.Logger = logger
return nil
}

Expand Down
7 changes: 5 additions & 2 deletions pkg/sink/kafka/sarama.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,15 +58,18 @@ func NewSaramaConfig(ctx context.Context, o *Options) (*sarama.Config, error) {
// For kafka cluster with a bad network condition, producer should not try to
// waster too much time on sending a message, get response no matter success
// or fail as soon as possible is preferred.
config.Producer.Retry.Max = 3
config.Producer.Retry.Backoff = 100 * time.Millisecond
// According to the https://github.com/IBM/sarama/issues/2619,
// sarama may send message out of order even set the `config.Net.MaxOpenRequest` to 1,
// when the kafka cluster is unhealthy and trigger the internal retry mechanism.
config.Producer.Retry.Max = 0

// make sure sarama producer flush messages as soon as possible.
config.Producer.Flush.Bytes = 0
config.Producer.Flush.Messages = 0
config.Producer.Flush.Frequency = time.Duration(0)
config.Producer.Flush.MaxMessages = o.MaxMessages

config.Net.MaxOpenRequests = 1
config.Net.DialTimeout = o.DialTimeout
config.Net.WriteTimeout = o.WriteTimeout
config.Net.ReadTimeout = o.ReadTimeout
Expand Down
Loading