Change kafka producer client config (#2030)

(cherry picked from commit 44054d4)

protocol/indexer/flags.go

@@ -16,12 +16,13 @@ type IndexerFlags struct {
 
 // List of default values
 const (
-	DefaultMaxRetries = 3
+	DefaultMaxRetries = 20
 )
 
 // List of CLI flags
 const (
-	FlagKafkaConnStr         = "indexer-kafka-conn-str"
+	FlagKafkaConnStr = "indexer-kafka-conn-str"
+	// max retry should be set so that max retry * retry backoff > Zookeeper session.timeout + some buffer
 	FlagKafkaMaxRetry        = "indexer-kafka-max-retry"
 	FlagSendOffchainData     = "indexer-send-offchain-data"
 	MsgSenderInstanceForTest = "msgsender-instance-for-test"

protocol/indexer/msgsender/msgsender_kafka.go

@@ -43,7 +43,10 @@ func NewIndexerMessageSenderKafka(
 	config.Producer.Return.Errors = true
 	config.Producer.Return.Successes = true
 	config.Producer.Retry.Max = indexerFlags.MaxRetries
+	// max retry should be set so that max retry * retry backoff > Zookeeper session.timeout + some buffer
+	config.Producer.Retry.Backoff = 1000 * time.Millisecond
 	config.Producer.MaxMessageBytes = 4194304 // 4MB
+	config.Producer.RequiredAcks = sarama.WaitForAll
 	// Use the JVM compatible parititoner to match `kafkajs` which is used in the indexer services.
 	config.Producer.Partitioner = kafkautil.NewJVMCompatiblePartitioner
 	producer, err := sarama.NewAsyncProducer(indexerFlags.KafkaAddrs, config)