Function TestFuncInitProducerId3

functional_producer_test.go:342–363

(t *testing.T)

}

func TestFuncInitProducerId3(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 = "TestFuncInitProducerId3"
	config.Producer.RequiredAcks = WaitForAll
	config.Producer.Retry.Max = 50
	config.Consumer.IsolationLevel = ReadCommitted
	config.Net.MaxOpenRequests = 1

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

	require.True(t, producer.(*asyncProducer).txnmgr.coordinatorSupportsBumpingEpoch)
}

type messageHandler struct {
	*testing.T