Function TestFuncIdempotentBufferedSequence

functional_producer_test.go:765–867

(t *testing.T)

}

func TestFuncIdempotentBufferedSequence(t *testing.T) {
	checkKafkaVersion(t, "0.11.0.0")
	setupFunctionalTest(t)
	defer teardownFunctionalTest(t)

	const (
		topic           = "test.1"
		partition int32 = 0
	)

	cfg := NewFunctionalTestConfig()
	cfg.Net.MaxOpenRequests = 1
	cfg.Producer.Idempotent = true
	cfg.Producer.RequiredAcks = WaitForAll
	cfg.Producer.Return.Successes = true
	cfg.Producer.Return.Errors = true
	cfg.Producer.Retry.Max = 64
	cfg.Producer.Retry.Backoff = 250 * time.Millisecond

	start := time.Now()

	producer, err := NewAsyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, cfg)
	require.NoError(t, err)
	defer producer.Close()

	asyncProd, ok := producer.(*asyncProducer)
	require.True(t, ok)

	waitForMessages := func(count int) {
		timeout := time.After(2 * time.Minute)
		for count > 0 {
			select {
			case <-timeout:
				t.Fatalf("timed out waiting for %d messages", count)
			case perr := <-producer.Errors():
				if perr != nil {
					t.Logf("producer error: %v", perr.Err)
				}
				count--
			case <-producer.Successes():
				count--
			}
		}
	}

	for i := 0; i < 5; i++ {
		producer.Input() <- &ProducerMessage{
			Topic:     topic,
			Partition: partition,
			Value:     StringEncoder(fmt.Sprintf("warmup-%d", i)),
		}
	}
	waitForMessages(5)

	leader, err := asyncProd.client.Leader(topic, partition)
	require.NoError(t, err)

	bp := asyncProd.getBrokerProducer(leader)