Function TestFuncProducingIdempotentWithBrokerFailure

functional_producer_test.go:689–763

(t *testing.T)

}

func TestFuncProducingIdempotentWithBrokerFailure(t *testing.T) {
	setupFunctionalTest(t)
	defer teardownFunctionalTest(t)

	config := NewFunctionalTestConfig()
	config.Producer.Flush.Frequency = 250 * time.Millisecond
	config.Producer.Idempotent = true
	config.Producer.Timeout = 500 * time.Millisecond
	config.Producer.Retry.Max = 1
	config.Producer.Retry.Backoff = 500 * time.Millisecond
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Producer.RequiredAcks = WaitForAll
	config.Net.MaxOpenRequests = 1

	producer, err := NewSyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, config)
	if err != nil {
		t.Fatal(err)
	}
	defer safeClose(t, producer)

	// Successfully publish a few messages
	for i := 0; i < 10; i++ {
		_, _, err = producer.SendMessage(&ProducerMessage{
			Topic: "test.1",
			Value: StringEncoder(fmt.Sprintf("%d message", i)),
		})
		if err != nil {
			t.Fatal(err)
		}
	}

	// break the brokers.
	for proxyName, proxy := range FunctionalTestEnv.Proxies {
		if !strings.Contains(proxyName, "kafka") {
			continue
		}
		if err := proxy.Disable(); err != nil {
			t.Fatal(err)
		}
	}

	// This should fail hard now
	for i := 10; i < 20; i++ {
		_, _, err = producer.SendMessage(&ProducerMessage{
			Topic: "test.1",
			Value: StringEncoder(fmt.Sprintf("%d message", i)),
		})
		if err == nil {
			t.Fatal(err)
		}
	}

	// Now bring the proxy back up
	for proxyName, proxy := range FunctionalTestEnv.Proxies {
		if !strings.Contains(proxyName, "kafka") {
			continue
		}