Function TestProducerRetryBufferLimits

async_producer_test.go:2737–2836

(t *testing.T)

}

func TestProducerRetryBufferLimits(t *testing.T) {
	broker := NewMockBroker(t, 1)
	defer broker.Close()
	topic := "test-topic"

	metadataRequestHandlerFunc := func(req *request) (res encoderWithHeader) {
		r := new(MetadataResponse)
		r.AddBroker(broker.Addr(), broker.BrokerID())
		r.AddTopicPartition(topic, 0, broker.BrokerID(), nil, nil, nil, ErrNoError)
		return r
	}

	produceRequestHandlerFunc := func(req *request) (res encoderWithHeader) {
		r := new(ProduceResponse)
		r.AddTopicPartition(topic, 0, ErrNotLeaderForPartition)
		return r
	}

	broker.SetHandlerFuncByMap(map[string]requestHandlerFunc{
		"ProduceRequest":  produceRequestHandlerFunc,
		"MetadataRequest": metadataRequestHandlerFunc,
	})

	tests := []struct {
		name            string
		configureBuffer func(*Config)
		messageSize     int
		numMessages     int
	}{
		{
			name: "MaxBufferLength",
			configureBuffer: func(config *Config) {
				config.Producer.Flush.MaxMessages = 1
				config.Producer.Retry.MaxBufferLength = minFunctionalRetryBufferLength
			},
			messageSize: 1, // Small message size
			numMessages: 10000,
		},
		{
			name: "MaxBufferBytes",
			configureBuffer: func(config *Config) {
				config.Producer.Flush.MaxMessages = 1
				config.Producer.Retry.MaxBufferBytes = minFunctionalRetryBufferBytes
			},
			messageSize: 950 * 1024, // 950 KB
			numMessages: 1000,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			config := NewTestConfig()
			config.Producer.Return.Successes = true
			tt.configureBuffer(config)

			producer, err := NewAsyncProducer([]string{broker.Addr()}, config)
			if err != nil {
				t.Fatal(err)