sink(ticdc): Set config.Metadata.Timeout correctly (#3665) (#3671)

cdc/sink/producer/kafka/kafka.go

@@ -537,17 +537,27 @@ func newSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 	}
 	config.Version = version
 	// See: https://kafka.apache.org/documentation/#replication
-	// When one of the brokers in a Kafka cluster is down, the partition leaders in this broker is broken, Kafka will election a new partition leader and replication logs, this process will last from a few seconds to a few minutes. Kafka cluster will not provide a writing service in this process.
-	// Time out in one minute(120 * 500ms).
+	// When one of the brokers in a Kafka cluster is down, the partition leaders
+	// in this broker is broken, Kafka will election a new partition leader and
+	// replication logs, this process will last from a few seconds to a few minutes.
+	// Kafka cluster will not provide a writing service in this process.
+	// Time out in one minute.
 	config.Metadata.Retry.Max = 120
 	config.Metadata.Retry.Backoff = 500 * time.Millisecond
+	// If it is not set, this means a metadata request against an unreachable
+	// cluster (all brokers are unreachable or unresponsive) can take up to
+	// `Net.[Dial|Read]Timeout * BrokerCount * (Metadata.Retry.Max + 1) +
+	// Metadata.Retry.Backoff * Metadata.Retry.Max`
+	// to fail.
+	// See: https://github.com/Shopify/sarama/issues/765
+	// and https://github.com/pingcap/ticdc/issues/3352.
+	config.Metadata.Timeout = 1 * time.Minute
 
 	config.Producer.Partitioner = sarama.NewManualPartitioner
 	config.Producer.MaxMessageBytes = c.MaxMessageBytes
 	config.Producer.Return.Successes = true
 	config.Producer.Return.Errors = true
 	config.Producer.RequiredAcks = sarama.WaitForAll
-
 	switch strings.ToLower(strings.TrimSpace(c.Compression)) {
 	case "none":
 		config.Producer.Compression = sarama.CompressionNone
@@ -580,7 +590,6 @@ func newSaramaConfig(ctx context.Context, c *Config) (*sarama.Config, error) {
 			return nil, errors.Trace(err)
 		}
 	}
-
 	if c.SaslScram != nil && len(c.SaslScram.SaslUser) != 0 {
 		config.Net.SASL.Enable = true
 		config.Net.SASL.User = c.SaslScram.SaslUser