Function TestTxnCanAbort

async_producer_test.go:2627–2735

(t *testing.T)

}

func TestTxnCanAbort(t *testing.T) {
	broker := NewMockBroker(t, 1)
	defer broker.Close()

	config := NewTestConfig()
	config.Producer.Idempotent = true
	config.Producer.Transaction.ID = "test"
	config.Version = V0_11_0_0
	config.Producer.RequiredAcks = WaitForAll
	config.Producer.Return.Errors = false
	config.Producer.Return.Successes = true
	config.Producer.Retry.Backoff = 0
	config.Producer.Flush.Messages = 1
	config.Producer.Retry.Max = 1
	config.Net.MaxOpenRequests = 1

	var (
		mu                   sync.Mutex
		addPartitionsCount   int
		produceRequestsCount int
	)

	broker.SetHandlerFuncByMap(map[string]requestHandlerFunc{
		"MetadataRequest": func(req *request) encoderWithHeader {
			resp := new(MetadataResponse)
			resp.Version = 4
			resp.ControllerID = broker.BrokerID()
			resp.AddBroker(broker.Addr(), broker.BrokerID())
			resp.AddTopic("test-topic", ErrNoError)
			resp.AddTopic("test-topic-2", ErrNoError)
			resp.AddTopicPartition("test-topic", 0, broker.BrokerID(), nil, nil, nil, ErrNoError)
			resp.AddTopicPartition("test-topic-2", 0, broker.BrokerID(), nil, nil, nil, ErrNoError)
			return resp
		},
		"FindCoordinatorRequest": func(req *request) encoderWithHeader {
			resp := new(FindCoordinatorResponse)
			resp.Version = 1
			resp.Coordinator = &Broker{id: broker.BrokerID(), addr: broker.Addr()}
			resp.Err = ErrNoError
			return resp
		},
		"InitProducerIDRequest": func(req *request) encoderWithHeader {
			return &InitProducerIDResponse{
				Err:           ErrNoError,
				ProducerID:    1,
				ProducerEpoch: 0,
			}
		},
		"AddPartitionsToTxnRequest": func(req *request) encoderWithHeader {
			mu.Lock()
			addPartitionsCount++
			count := addPartitionsCount
			mu.Unlock()

			if count == 1 {
				return &AddPartitionsToTxnResponse{
					Errors: map[string][]*PartitionError{
						"test-topic-2": {{Partition: 0, Err: ErrNoError}},