Function TestProduceSetV3RequestBuilding

produce_set_test.go:221–288

(t *testing.T)

}

func TestProduceSetV3RequestBuilding(t *testing.T) {
	parent, ps := makeProduceSet()
	parent.conf.Producer.RequiredAcks = WaitForAll
	parent.conf.Producer.Timeout = 10 * time.Second
	parent.conf.Version = V0_11_0_0

	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,
	}
	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.MaxTimestamp.Equal(now.Add(9 * time.Second).Truncate(time.Millisecond)) {
		t.Errorf("Wrong max timestamp: %v", batch.MaxTimestamp)
	}
	for i := range 10 {
		rec := batch.Records[i]
		if rec.TimestampDelta != time.Duration(i)*time.Second {
			t.Errorf("Wrong timestamp delta: %v", rec.TimestampDelta)
		}

		if rec.OffsetDelta != int64(i) {
			t.Errorf("Wrong relative inner offset, expected %d, got %d", i, rec.OffsetDelta)
		}

		for j, h := range batch.Records[i].Headers {
			exp := fmt.Sprintf("header-%d", j+1)