Function TestProduceSetIdempotentRequestBuilding

produce_set_test.go:290–376

(t *testing.T)

}

func TestProduceSetIdempotentRequestBuilding(t *testing.T) {
	const pID = 1000
	const pEpoch = 1234

	config := NewTestConfig()
	config.Producer.RequiredAcks = WaitForAll
	config.Producer.Idempotent = true
	config.Version = V0_11_0_0

	parent := &asyncProducer{
		conf: config,
		txnmgr: &transactionManager{
			producerID:    pID,
			producerEpoch: pEpoch,
		},
	}
	ps := newProduceSet(parent)

	now := time.Now()
	msg := &ProducerMessage{
		Topic:     "t1",
		Partition: 0,
		Key:       StringEncoder(TestMessage),
		Value:     StringEncoder(TestMessage),
		Headers: []RecordHeader{
			{
				Key:   []byte("header-1"),
				Value: []byte("value-1"),
			},
			{
				Key:   []byte("header-2"),
				Value: []byte("value-2"),
			},
			{
				Key:   []byte("header-3"),
				Value: []byte("value-3"),
			},
		},
		Timestamp:      now,
		sequenceNumber: 123,
	}
	for range 10 {
		safeAddMessage(t, ps, msg)
		msg.Timestamp = msg.Timestamp.Add(time.Second)
	}

	req := ps.buildRequest()

	if req.Version != 3 {
		t.Error("Wrong request version")
	}

	batch := req.records["t1"][0].RecordBatch
	if !batch.FirstTimestamp.Equal(now.Truncate(time.Millisecond)) {
		t.Errorf("Wrong first timestamp: %v", batch.FirstTimestamp)
	}
	if batch.ProducerID != pID {
		t.Errorf("Wrong producerID: %v", batch.ProducerID)