Function TestFuncTxnProduceEpochBump

functional_producer_test.go:274–340

(t *testing.T)

}

func TestFuncTxnProduceEpochBump(t *testing.T) {
	checkKafkaVersion(t, "2.6.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 = "TestFuncTxnProduceEpochBump"
	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()

	err = producer.BeginTxn()
	require.NoError(t, 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)

	for i := 0; i < 1; i++ {
		msg := <-msgChannel
		t.Logf("Received %s from %s-%d at offset %d", msg.Value, msg.Topic, msg.Partition, msg.Offset)
	}

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

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