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) #11962

Merged
merged 3 commits into from
Jan 7, 2025
Merged

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

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
3797961
set the sarama producer retry to 0
3AceShowHand Dec 12, 2024
eda0b21
enable the sarama log
3AceShowHand Dec 26, 2024
2266929
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 @@

// 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)

Check warning on line 228 in pkg/logutil/log.go

View check run for this annotation

Codecov / codecov/patch

pkg/logutil/log.go#L226-L228 

Added lines #L226 - L228 were not covered by tests
}
sarama.Logger = logger

Check warning on line 230 in pkg/logutil/log.go

View check run for this annotation

Codecov / codecov/patch

pkg/logutil/log.go#L230 

Added line #L230 was not covered by tests
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