Function TestTxnmgrInitProducerIdTxnCoordinatorLoading

transaction_manager_test.go:132–176

TestTxnmgrInitProducerIdTxnCoordinatorLoading ensure we retry initProducerId when either FindCoordinator or InitProducerID returns ErrOffsetsLoadInProgress

(t *testing.T)

// TestTxnmgrInitProducerIdTxnCoordinatorLoading ensure we retry initProducerId when either FindCoordinator or InitProducerID returns ErrOffsetsLoadInProgress
func TestTxnmgrInitProducerIdTxnCoordinatorLoading(t *testing.T) {
	config := NewTestConfig()
	config.Producer.Idempotent = true
	config.Producer.Transaction.ID = "txid-group"
	config.Version = V0_11_0_0
	config.Producer.RequiredAcks = WaitForAll
	config.Net.MaxOpenRequests = 1

	broker := NewMockBroker(t, 1)
	defer broker.Close()

	broker.SetHandlerByMap(map[string]MockResponse{
		"MetadataRequest": NewMockMetadataResponse(t).
			SetController(broker.BrokerID()).
			SetBroker(broker.Addr(), broker.BrokerID()),
		"FindCoordinatorRequest": NewMockSequence(
			NewMockFindCoordinatorResponse(t).
				SetError(CoordinatorTransaction, "txid-group", ErrOffsetsLoadInProgress),
			NewMockFindCoordinatorResponse(t).
				SetError(CoordinatorTransaction, "txid-group", ErrOffsetsLoadInProgress),
			NewMockFindCoordinatorResponse(t).
				SetCoordinator(CoordinatorTransaction, "txid-group", broker),
		),
		"InitProducerIDRequest": NewMockSequence(
			NewMockInitProducerIDResponse(t).
				SetError(ErrOffsetsLoadInProgress),
			NewMockInitProducerIDResponse(t).
				SetError(ErrOffsetsLoadInProgress),
			NewMockInitProducerIDResponse(t).
				SetProducerID(1).
				SetProducerEpoch(0),
		),
	})

	client, err := NewClient([]string{broker.Addr()}, config)
	require.NoError(t, err)
	defer client.Close()

	txmng, err := newTransactionManager(config, client)
	require.NoError(t, err)

	require.Equal(t, int64(1), txmng.producerID)
	require.Equal(t, int16(0), txmng.producerEpoch)
	require.Equal(t, ProducerTxnFlagReady, txmng.status)
}

func TestMaybeAddPartitionToCurrentTxn(t *testing.T) {
	type testCase struct {