Function testTxnProduceAndCommitOffset

functional_producer_test.go:417–514

(
	t *testing.T,
	version KafkaVersion,
	txnID, groupID string,
	addOffsetToTxn func(producer AsyncProducer, sess ConsumerGroupSession, msg *ConsumerMessage, groupID string) error,
)

}

func testTxnProduceAndCommitOffset(
	t *testing.T,
	version KafkaVersion,
	txnID, groupID string,
	addOffsetToTxn func(producer AsyncProducer, sess ConsumerGroupSession, msg *ConsumerMessage, groupID string) error,
) {
	setupFunctionalTest(t)
	defer teardownFunctionalTest(t)

	config := NewFunctionalTestConfig()
	config.Version = version
	config.ChannelBufferSize = 20
	config.Producer.Flush.Frequency = 50 * time.Millisecond
	config.Producer.Flush.Messages = 200
	config.Producer.Idempotent = true
	config.Producer.Transaction.ID = txnID
	config.Producer.RequiredAcks = WaitForAll
	config.Producer.Transaction.Retry.Max = 200
	config.Consumer.IsolationLevel = ReadCommitted
	config.Consumer.Offsets.AutoCommit.Enable = false
	config.Net.MaxOpenRequests = 1

	client, err := NewClient(FunctionalTestEnv.KafkaBrokerAddrs, config)
	require.NoError(t, err)
	defer client.Close()

	admin, err := NewClusterAdminFromClient(client)
	require.NoError(t, err)
	defer admin.Close()

	producer, err := NewAsyncProducerFromClient(client)
	require.NoError(t, err)
	defer producer.Close()

	cg, err := NewConsumerGroupFromClient(groupID, client)
	require.NoError(t, err)
	defer cg.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	handler := &messageHandler{started: make(chan struct{})}
	handler.T = t
	handler.h = func(sess ConsumerGroupSession, msg *ConsumerMessage) {
		require.NoError(t, producer.BeginTxn())
		producer.Input() <- &ProducerMessage{Topic: "test.1", Value: StringEncoder("test-prod")}
		require.NoError(t, addOffsetToTxn(producer, sess, msg, groupID))
		require.NoError(t, producer.CommitTxn())
	}

	go func() {
		err = cg.Consume(ctx, []string{"test.4"}, handler)
		assert.NoError(t, err)
	}()

	<-handler.started

	nonTransactionalProducer, err := NewAsyncProducer(FunctionalTestEnv.KafkaBrokerAddrs, NewFunctionalTestConfig())