Function TestFuncTxnProduceWithBrokerFailure

functional_producer_test.go:209–272

(t *testing.T)

}

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

	config := NewFunctionalTestConfig()
	config.ChannelBufferSize = 20
	config.Producer.Flush.Frequency = 50 * time.Millisecond
	config.Producer.Flush.Messages = 200
	config.Producer.Idempotent = true
	config.Producer.Transaction.ID = "TestFuncTxnProduceWithBrokerFailure"
	config.Producer.RequiredAcks = WaitForAll
	config.Producer.Transaction.Retry.Max = 200
	config.Consumer.IsolationLevel = ReadCommitted
	config.Net.MaxOpenRequests = 1

	consumer, err := NewConsumer(FunctionalTestEnv.KafkaBrokerAddrs, config)
	require.NoError(t, err)
	defer consumer.Close()

	pc, err := consumer.ConsumePartition("test.1", 0, OffsetNewest)
	require.NoError(t, err)
	msgChannel := pc.Messages()
	defer pc.Close()

	nonTransactionalProducer, err := NewAsyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, NewFunctionalTestConfig())
	require.NoError(t, err)
	defer nonTransactionalProducer.Close()

	// Ensure consumer is started
	nonTransactionalProducer.Input() <- &ProducerMessage{Topic: "test.1", Key: nil, Value: StringEncoder("test")}
	<-msgChannel

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

	txCoordinator, _ := producer.(*asyncProducer).client.TransactionCoordinator(config.Producer.Transaction.ID)

	err = producer.BeginTxn()
	require.NoError(t, err)

	defer func() {
		if err := startDockerTestBroker(context.Background(), txCoordinator.id); err != nil {
			t.Fatal(err)
		}
		t.Logf("\n")
	}()
	if err := stopDockerTestBroker(context.Background(), txCoordinator.id); err != nil {
		t.Fatal(err)
	}

	for i := 0; i < 1; i++ {
		producer.Input() <- &ProducerMessage{Topic: "test.1", Key: nil, Value: StringEncoder("test")}
	}

	err = producer.CommitTxn()
	require.NoError(t, err)