Function TestAsyncProducerRetryOrdering

async_producer_test.go:2927–3049

TestAsyncProducerRetryOrdering verifies that message ordering is preserved during retries, both with and without request pipelining (MaxOpenRequests=1 vs >1).

(t *testing.T)

// TestAsyncProducerRetryOrdering verifies that message ordering is preserved during retries,
// both with and without request pipelining (MaxOpenRequests=1 vs >1).
func TestAsyncProducerRetryOrdering(t *testing.T) {
	const topic = "my_topic"

	extractValue := func(pr *ProduceRequest) string {
		recordsByPartition := pr.records[topic]
		if recordsByPartition == nil {
			return ""
		}
		records := recordsByPartition[0]
		if rb := records.RecordBatch; rb != nil && len(rb.Records) > 0 {
			return string(rb.Records[0].Value)
		}
		if ms := records.MsgSet; ms != nil && len(ms.Messages) > 0 {
			return string(ms.Messages[0].Msg.Value)
		}
		return ""
	}

	tests := []struct {
		name            string
		maxOpenRequests int
		retryBackoff    time.Duration
	}{
		{
			name:            "no pipelining (MaxOpenRequests=1)",
			maxOpenRequests: 1,
			retryBackoff:    0,
		},
		{
			name:            "with pipelining (MaxOpenRequests=5)",
			maxOpenRequests: 5,
			retryBackoff:    50 * time.Millisecond,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			seedBroker := NewMockBroker(t, 1)
			leader := NewMockBroker(t, 2)

			metadataResponse := new(MetadataResponse)
			metadataResponse.AddBroker(leader.Addr(), leader.BrokerID())
			metadataResponse.AddTopicPartition(topic, 0, leader.BrokerID(), nil, nil, nil, ErrNoError)
			seedBroker.Returns(metadataResponse)

			var (
				mu              sync.Mutex
				produceAttempts int
				valuesSeen      []string
			)

			leader.setHandler(func(req *request) (res encoderWithHeader) {
				switch typed := req.body.(type) {
				case *MetadataRequest:
					return metadataResponse
				case *ProduceRequest:
					mu.Lock()
					defer mu.Unlock()