Multipartition topic produce test

functional_test_promise.go

func TestFuncMultiPartitionProduce(t *testing.T) {
	checkKafkaAvailability(t)
	Logger = log.New(os.Stderr, "[Sarama] ", log.Lmicroseconds|log.Lshortfile)
	client, err := NewClient("functional_test", []string{kafkaAddr}, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer safeClose(t, client)

	config := newProdConf()
	producer, err := NewProducer(client, config)
	if err != nil {
		t.Fatal(err)
	}

	var wg sync.WaitGroup
	wg.Add(TestBatchSize)

	for i := 1; i <= TestBatchSize; i++ {

		go func(i int, w *sync.WaitGroup) {
			defer w.Done()
			promise := make(chan error)
			defer close(promise)
			msg := &MessageToSend{Topic: "multi_partition", Key: nil, Value: StringEncoder(fmt.Sprintf("hur %d", i)), Promise: promise}
			producer.Input() <- msg
			select {
			case ret := <-producer.Errors():
				t.Fatal(ret.Err)
			case err := <-promise:
				if err != nil {
					t.Fatal(err)
				}
			}
		}(i, &wg)
	}

	wg.Wait()
	if err := producer.Close(); err != nil {
		t.Error(err)
	}
}

func newProdConf() *ProducerConfig {
	pc := NewProducerConfig()
	pc.FlushFrequency = 50 * time.Millisecond
	pc.FlushMsgCount = 200
	pc.ChannelBufferSize = 20
	return pc
}