func SetupProducer(brokers string, serializer Serializer) (Producer, error) {
	brokerList := strings.Split(brokers, ",")
	if len(brokerList) == 0 {
		return nil, errors.New("The Kafka producer requires at least 1 broker to resolve topics.")
	}

	saramaConfig := sarama.NewConfig()
	saramaConfig.Net.TLS.Enable = false
	saramaConfig.Producer.RequiredAcks = sarama.WaitForAll // Wait for all in-sync replicas to ack the message
	saramaConfig.Producer.Retry.Max = 10
	saramaConfig.Producer.Return.Errors = true
	saramaConfig.Producer.Return.Successes = true
	// saramaConfig.Producer.Flush.Frequency = 50 * time.Millisecond
	saramaConfig.Producer.Partitioner = partitioner.NewMurmur2Partitioner

	saramaProducer, err := sarama.NewSyncProducer(brokerList, saramaConfig)
	if err != nil {
		return nil, err
	}

	return &producerImp{saramaProducer: saramaProducer, serializer: serializer}, nil
}
// mustSetupProducer creates a producer for the kafka queues on brokers
// brokers is a comma separated list of kafka nodes with host and ip, eg "172.0.0.1:9092,172.0.0.2:9092"
mustSetupProducer creates a producer for the kafka queues on brokers
brokers is a comma separated list of kafka nodes with host and ip, eg "172.0.0.1:9092,172.0.0.2:9092"
// mustSetupProducer creates a producer for the kafka queues on brokers // brokers is a comma separated list of kafka nodes with host and ip eg " 172 . 0 . 0 . 1 : 9092 172 . 0 . 0 . 2 : 9092 "
func SetupProducer(brokers string, serializer Serializer) (Producer, error) {
	brokerList := strings.Split(brokers, ",")
	if len(brokerList) == 0 {
		return nil, errors.New("The Kafka producer requires at least 1 broker to resolve topics.")
	}
	saramaConfig := sarama.NewConfig()
	saramaConfig.Net.TLS.Enable = false
	saramaConfig.Producer.RequiredAcks = sarama.WaitForAll 
	saramaConfig.Producer.Retry.Max = 10
	saramaConfig.Producer.Return.Errors = true
	saramaConfig.Producer.Return.Successes = true
	saramaConfig.Producer.Partitioner = partitioner.NewMurmur2Partitioner
	saramaProducer, err := sarama.NewSyncProducer(brokerList, saramaConfig)
	if err != nil {
		return nil, err
	}
	return &producerImp{saramaProducer: saramaProducer, serializer: serializer}, nil
}
func SetupProducer ( brokers string , serializer Serializer ) ( Producer , error ) { brokerList := strings . Split ( brokers , " " ) if len ( brokerList ) == 0 { return nil , errors . New ( " " ) } saramaConfig := sarama . NewConfig ( ) saramaConfig . Net . TLS . Enable = false saramaConfig . Producer . RequiredAcks = sarama . WaitForAll saramaConfig . Producer . Retry . Max = 10 saramaConfig . Producer . Return . Errors = true saramaConfig . Producer . Return . Successes = true saramaConfig . Producer . Partitioner = partitioner . NewMurmur2Partitioner saramaProducer , err := sarama . NewSyncProducer ( brokerList , saramaConfig ) if err != nil { return nil , err } return & producerImp { saramaProducer : saramaProducer , serializer : serializer } , nil }
mustSetupProducer creates a producer for the kafka queues on brokers
brokers is a comma separated list of kafka nodes with host and ip, eg "172.0.0.1:9092,172.0.0.2:9092"
mustSetupProducer creates a producer for the kafka queues on brokers brokers is a comma separated list of kafka nodes with host and ip eg " 172 . 0 . 0 . 1 : 9092 172 . 0 . 0 . 2 : 9092 "
// Wait for all in-sync replicas to ack the message // saramaConfig.Producer.Flush.Frequency = 50 * time.Millisecond