Function newMockTxnManager

transaction_manager_test.go:446–465

newMockTxnManager wires a transaction manager to broker using the standard metadata, find-coordinator and init-producer-id handshake. The caller owns closing the returned client.

(t *testing.T, broker *MockBroker, config *Config)

// metadata, find-coordinator and init-producer-id handshake. The caller owns
// closing the returned client.
func newMockTxnManager(t *testing.T, broker *MockBroker, config *Config) (Client, *transactionManager) {
	// the mock broker matches each response version to the request version, so
	// MetadataResponse.Version is left at its zero value here
	metadataLeader := new(MetadataResponse)
	metadataLeader.ControllerID = broker.brokerID
	metadataLeader.AddBroker(broker.Addr(), broker.BrokerID())
	metadataLeader.AddTopic("test-topic", ErrNoError)
	metadataLeader.AddTopicPartition("test-topic", 0, broker.BrokerID(), nil, nil, nil, ErrNoError)
	broker.Returns(metadataLeader)

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

	broker.Returns(&FindCoordinatorResponse{Coordinator: client.Brokers()[0], Err: ErrNoError})
	broker.Returns(&InitProducerIDResponse{Err: ErrNoError, ProducerID: 1, ProducerEpoch: 0})

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

func TestTxnOffsetCommitGroupMetadata(t *testing.T) {
	broker := NewMockBroker(t, 1)