Function TestFuncTxnAbortedProduce

functional_producer_test.go:599–669

(t *testing.T)

}

func TestFuncTxnAbortedProduce(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 = "TestFuncTxnAbortedProduce"
	config.Producer.RequiredAcks = WaitForAll
	config.Producer.Return.Successes = true
	config.Producer.Transaction.Retry.Max = 200
	config.Consumer.IsolationLevel = ReadCommitted
	config.Net.MaxOpenRequests = 1

	client, err := NewClient(FunctionalTestEnv.KafkaBrokerAddrs, config)
	require.NoError(t, err)
	defer client.Close()

	consumer, err := NewConsumerFromClient(client)
	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()

	nonTransactionalConfig := NewFunctionalTestConfig()
	nonTransactionalConfig.Producer.Return.Successes = true
	nonTransactionalConfig.Producer.Return.Errors = true

	nonTransactionalProducer, err := NewAsyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, nonTransactionalConfig)
	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 := NewAsyncProducerFromClient(client)
	require.NoError(t, err)
	defer producer.Close()

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

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

	for i := 0; i < 2; i++ {
		<-producer.Successes()
	}

	err = producer.AbortTxn()