adjust layout for sarama config producer.

pingcap · Nov 8, 2021 · 1796153 · 1796153
1 parent 5635eff
commit 1796153
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/cdc/sink/producer/kafka/kafka.go b/cdc/sink/producer/kafka/kafka.go
@@ -600,6 +600,10 @@ func newSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 	config.Producer.Return.Errors = true
 	config.Producer.RequiredAcks = sarama.WaitForAll
 
+	// Time out in five minutes(600 * 500ms).
+	config.Producer.Retry.Max = 600
+	config.Producer.Retry.Backoff = 500 * time.Millisecond
+
 	switch strings.ToLower(strings.TrimSpace(c.Compression)) {
 	case "none":
 		config.Producer.Compression = sarama.CompressionNone
@@ -616,10 +620,6 @@ func newSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 		config.Producer.Compression = sarama.CompressionNone
 	}
 
-	// Time out in five minutes(600 * 500ms).
-	config.Producer.Retry.Max = 600
-	config.Producer.Retry.Backoff = 500 * time.Millisecond
-
 	// Time out in one minute(120 * 500ms).
 	config.Admin.Retry.Max = 120
 	config.Admin.Retry.Backoff = 500 * time.Millisecond