func initProducer(moduleConfig *Config) (*Producer, error) {
	fmt.Println("[INFO] initProducer called")
	brokerList := moduleConfig.Kafka.BrokerList
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll // only wait for leader to ack
	config.Producer.Compression = sarama.CompressionSnappy
	config.Producer.Flush.Frequency = 500 * time.Millisecond

	var producer sarama.AsyncProducer
	var err error
	for currConnAttempt := 0; currConnAttempt < moduleConfig.Kafka.MaxRetry; currConnAttempt++ {
		producer, err = sarama.NewAsyncProducer(brokerList, config)
		if err == nil {
			break
		}
		fmt.Println("[INFO] Connection attempt faild (", (currConnAttempt + 1), "/", moduleConfig.Kafka.MaxRetry, ")")
		<-time.After(time.Second * 5)
	}

	if err != nil {
		fmt.Println("[ERROR] Unable to setup kafka producer", err)
		return nil, err
	}

	//You must read from the Errors() channel or the producer will deadlock.
	go func() {
		for err := range producer.Errors() {
			log.Println("[ERROR] Kadka producer Error: ", err)
		}
	}()

	fmt.Println("[INFO] kafka producer initialized successfully")
	return &Producer{producer: producer, id: CreatedProducersLength()}, nil
}

func (this *Mirror) pump(sub *consumergroup.ConsumerGroup, pub sarama.AsyncProducer, stop chan struct{}) {
	defer func() {
		log.Println("pump cleanup...")
		sub.Close()
		log.Println("pump cleanup ok")

		stop <- struct{}{} // notify others I'm done
	}()

	log.Printf("start pumping")
	active := false
	for {
		select {
		case <-this.quit:
			return

		case <-stop:
			// yes sir!
			return

		case <-time.After(time.Second * 10):
			active = false
			log.Println("idle 10s waiting for new msg")

		case msg := <-sub.Messages():
			if !active || this.debug {
				log.Printf("<-[%d] T:%s M:%s", this.transferN, msg.Topic, string(msg.Value))
			}
			active = true

			pub.Input() <- &sarama.ProducerMessage{
				Topic: msg.Topic,
				Key:   sarama.ByteEncoder(msg.Key),
				Value: sarama.ByteEncoder(msg.Value),
			}
			if this.autoCommit {
				sub.CommitUpto(msg)
			}

			// rate limit, never overflood the limited bandwidth between IDCs
			// FIXME when compressed, the bandwidth calculation is wrong
			bytesN := len(msg.Topic) + len(msg.Key) + len(msg.Value) + 20 // payload overhead
			if !this.bandwidthRateLimiter.Pour(bytesN) {
				time.Sleep(time.Second)
				this.Ui.Warn(fmt.Sprintf("%d -> bandwidth reached, backoff 1s", bytesN))
			}
			this.transferBytes += int64(bytesN)

			this.transferN++
			if this.transferN%this.progressStep == 0 {
				log.Println(gofmt.Comma(this.transferN))
			}

		case err := <-sub.Errors():
			this.Ui.Error(err.Error()) // TODO
		}
	}
}

func (suite *KafkaTester) Test01() {
	t := suite.T()
	assert := assert.New(t)

	const M1 = "message one"
	const M2 = "message two"

	var producer sarama.AsyncProducer
	var consumer sarama.Consumer
	var partitionConsumer sarama.PartitionConsumer

	var err error

	topic := makeTopicName()

	{
		config := sarama.NewConfig()
		config.Producer.Return.Successes = false
		config.Producer.Return.Errors = false

		producer, err = sarama.NewAsyncProducer([]string{suite.server}, config)
		assert.NoError(err)
		defer close(t, producer)

		producer.Input() <- &sarama.ProducerMessage{
			Topic: topic,
			Key:   nil,
			Value: sarama.StringEncoder(M1)}

		producer.Input() <- &sarama.ProducerMessage{
			Topic: topic,
			Key:   nil,
			Value: sarama.StringEncoder(M2)}
	}

	{
		consumer, err = sarama.NewConsumer([]string{suite.server}, nil)
		assert.NoError(err)
		defer close(t, consumer)

		partitionConsumer, err = consumer.ConsumePartition(topic, 0, 0)
		assert.NoError(err)
		defer close(t, partitionConsumer)
	}

	{
		mssg1 := <-partitionConsumer.Messages()
		//t.Logf("Consumed: offset:%d  value:%v", mssg1.Offset, string(mssg1.Value))
		mssg2 := <-partitionConsumer.Messages()
		//t.Logf("Consumed: offset:%d  value:%v", mssg2.Offset, string(mssg2.Value))

		assert.EqualValues(M1, string(mssg1.Value))
		assert.EqualValues(M2, string(mssg2.Value))
	}
}

func Publish(input chan *FileEvent, source string, ctrl chan bool) {
	clientConfig := sarama.NewConfig()
	clientConfig.Producer.RequiredAcks = sarama.WaitForLocal
	clientConfig.Producer.Compression = sarama.CompressionSnappy
	clientConfig.Producer.Flush.Frequency = 500 * time.Millisecond
	clientConfig.Producer.Flush.Messages = 200
	clientConfig.Producer.Flush.MaxMessages = 200
	clientConfig.Producer.Flush.Bytes = 16384
	clientConfig.Producer.Return.Successes = true
	clientConfig.Producer.Partitioner = sarama.NewRoundRobinPartitioner
	clientConfig.ChannelBufferSize = kafkabuffer

	//brokerList := []string{"127.0.0.1:9092"}
	var producer sarama.AsyncProducer
	var err error
	for {
		producer, err = sarama.NewAsyncProducer(brokerList, clientConfig)
		if err != nil {
			log.Error("Publish: Failed to start Sarama producer: ", err)
			log.Info("waiting....")
			time.Sleep(1 * time.Second)
		} else {
			break
		}
	}

	defer func() {
		if err := producer.Close(); err != nil {
			log.Error("Failed to shutdown producer cleanly", err)
		}
	}()

	registrar := &Registrar{source: source, publishCtrl: ctrl}
	go registrar.RegistrarDo(producer.Errors(), producer.Successes())

	topic := kafkaTopic
	baseName := filepath.Base(source)
	if len(topicmap) > 0 {
		tmpTopic := genTopic(baseName, topicmap)
		if tmpTopic != "" {
			topic = tmpTopic
		}
	}

	key := hashKey
	for event := range input {
		log.Debugf("%v, %v, %v, %v", *event.Source, *event.Text, event.Line, event.Offset)
		key = strconv.FormatInt(event.Offset, 10)
		producer.Input() <- &sarama.ProducerMessage{
			Topic:    topic,
			Key:      sarama.StringEncoder(key),
			Value:    sarama.StringEncoder(*event.Text),
			Metadata: event,
		}
	}

}

func emitPacket(pckt Packt, producer sarama.AsyncProducer) {
	// Serialize the packet struct to JSON and send
	// to kafka topic
	var json, err = json.Marshal(pckt)
	if err == nil {
		producer.Input() <- &sarama.ProducerMessage{
			Topic: "packetstorm",
			Value: sarama.ByteEncoder(json),
		}
	} else {
		log.Fatal("Couldn't marshall the packet data")
	}
}

func expectationProducer(p sarama.AsyncProducer, expectations chan<- *sarama.ProducerMessage, wg *sync.WaitGroup) {
	defer wg.Done()

	var producerWg sync.WaitGroup

	producerWg.Add(1)
	go func() {
		defer producerWg.Done()
		for msg := range p.Successes() {
			stats.LogProduced(msg)
			expectations <- msg
		}
	}()

	producerWg.Add(1)
	go func() {
		defer producerWg.Done()
		for err := range p.Errors() {
			logger.Println("Failed to produce message:", err)
		}
	}()

	go monitor()
	logger.Printf("Producing %d messages...\n", *batchSize)

ProducerLoop:
	for i := 0; i < *batchSize; i++ {
		msg := &sarama.ProducerMessage{
			Topic:    *topic,
			Key:      sarama.StringEncoder(fmt.Sprintf("%d", i)),
			Value:    nil,
			Metadata: &MessageMetadata{Enqueued: time.Now()},
		}

		select {
		case <-shutdown:
			logger.Println("Early shutdown initiated...")
			break ProducerLoop
		case p.Input() <- msg:
			stats.LogEnqueued(msg)
		}

		if *sleep > 0 {
			time.Sleep(time.Duration(*sleep))
		}
	}

	p.AsyncClose()
	producerWg.Wait()
	close(expectations)
}

func consumer(src <-chan uint32, dst <-chan uint32, producer kafka.AsyncProducer) {
	topic := "network"

	db, err := geoip.Open("GeoLite2-City.mmdb")
	if err != nil {
		fmt.Println(err)
	}
	defer db.Close()

	for {
		src_ip := <-src
		src_str := intToIP(src_ip)
		dst_ip := <-dst
		dst_str := intToIP(dst_ip)

		src_ip_parsed := net.ParseIP(src_str)
		dst_ip_parsed := net.ParseIP(dst_str)
		src_record, err := db.City(src_ip_parsed)
		if err != nil {
			fmt.Println(err)
		}
		dst_record, err := db.City(dst_ip_parsed)
		if err != nil {
			fmt.Println(err)
		}
		src_coords := fmt.Sprintf("[%v,%v]", src_record.Location.Latitude, src_record.Location.Longitude)
		dst_coords := fmt.Sprintf("[%v,%v]", dst_record.Location.Latitude, dst_record.Location.Longitude)
		t := time.Now()
		current_time := fmt.Sprintf("%d-%02d-%02dT%02d:%02d:%02d-00:00",
			t.Year(), t.Month(), t.Day(),
			t.Hour(), t.Minute(), t.Second())
		if (src_coords == "0,0") || dst_coords == "0,0" {
			continue
		}
		str := src_str + "," + dst_str + "," + "\"" + src_coords + "\"" + "," + "\"" + dst_coords + "\"" + "," + current_time
		fmt.Println(str)

		message := kafka.ProducerMessage{Topic: topic, Value: kafka.StringEncoder(str)}
		producer.Input() <- &message
	}
}

func (this *Mirror) pump(sub *consumergroup.ConsumerGroup, pub sarama.AsyncProducer,
	stop, stopped chan struct{}) {
	defer func() {
		log.Trace("closing sub, commit offsets...")
		sub.Close()

		stopped <- struct{}{} // notify others I'm done
	}()

	active := true
	backoff := time.Second * 2
	idle := time.Second * 10
	for {
		select {
		case <-this.quit:
			log.Trace("got signal quit")
			return

		case <-stop:
			// yes sir!
			log.Trace("got signal stop")
			return

		case <-time.After(idle):
			active = false
			log.Info("idle 10s waiting for new message")

		case msg, ok := <-sub.Messages():
			if !ok {
				log.Warn("sub encounters end of message stream")
				return
			}

			if !active || this.Debug {
				log.Info("<-[#%d] T:%s M:%s", this.transferN, msg.Topic, string(msg.Value))
			}
			active = true

			pub.Input() <- &sarama.ProducerMessage{
				Topic: msg.Topic,
				Key:   sarama.ByteEncoder(msg.Key),
				Value: sarama.ByteEncoder(msg.Value),
			}
			if this.AutoCommit {
				sub.CommitUpto(msg)
			}

			// rate limit, never overflood the limited bandwidth between IDCs
			// FIXME when compressed, the bandwidth calculation is wrong
			bytesN := len(msg.Topic) + len(msg.Key) + len(msg.Value) + 20 // payload overhead
			if this.bandwidthRateLimiter != nil && !this.bandwidthRateLimiter.Pour(bytesN) {
				log.Warn("%s -> bandwidth reached, backoff %s", gofmt.ByteSize(this.transferBytes), backoff)
				time.Sleep(backoff)
			}

			this.transferBytes += int64(bytesN)
			this.transferN++
			if this.transferN%this.ProgressStep == 0 {
				log.Trace("%s %s %s", gofmt.Comma(this.transferN), gofmt.ByteSize(this.transferBytes), msg.Topic)
			}

		case err := <-sub.Errors():
			log.Error("quitting pump %v", err)
			return
		}
	}
}

func (this *SyslogProducer) produceRoutine(producer sarama.AsyncProducer) {
	for msg := range this.incoming {
		Tracef(this, "Got message: %s", msg)
		producer.Input() <- this.config.Transformer(msg, this.config.Topic)
	}
}