sarama/fetch_request_test.go

package sarama

import "testing"

var (
	fetchRequestNoBlocks = []byte{
		0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
		0x00, 0x00, 0x00, 0x00}

	fetchRequestWithProperties = []byte{
		0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x20, 0x00, 0x00, 0x00, 0xEF,
		0x00, 0x00, 0x00, 0x00}

	fetchRequestOneBlock = []byte{
		0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
		0x00, 0x00, 0x00, 0x01,
		0x00, 0x05, 't', 'o', 'p', 'i', 'c',
		0x00, 0x00, 0x00, 0x01,
		0x00, 0x00, 0x00, 0x12, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x34, 0x00, 0x00, 0x00, 0x56}

	fetchRequestOneBlockV4 = []byte{
		0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
		0x00, 0x00, 0x00, 0xFF,
		0x01,
		0x00, 0x00, 0x00, 0x01,
		0x00, 0x05, 't', 'o', 'p', 'i', 'c',
		0x00, 0x00, 0x00, 0x01,
		0x00, 0x00, 0x00, 0x12, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x34, 0x00, 0x00, 0x00, 0x56}

	fetchRequestOneBlockV11 = []byte{
		0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
		0x00, 0x00, 0x00, 0xFF,
		0x01,
		0x00, 0x00, 0x00, 0xAA, // sessionID
		0x00, 0x00, 0x00, 0xEE, // sessionEpoch
		0x00, 0x00, 0x00, 0x01,
		0x00, 0x05, 't', 'o', 'p', 'i', 'c',
		0x00, 0x00, 0x00, 0x01,
		0x00, 0x00, 0x00, 0x12, // partitionID
		0xFF, 0xFF, 0xFF, 0xFF, // currentLeaderEpoch
		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x34, // fetchOffset
		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // logStartOffset
		0x00, 0x00, 0x00, 0x56, // maxBytes
		0x00, 0x00, 0x00, 0x00,
		0x00, 0x06, 'r', 'a', 'c', 'k', '0', '1', // rackID
	}
)

func TestFetchRequest(t *testing.T) {
	t.Run("no blocks", func(t *testing.T) {
		request := new(FetchRequest)
		testRequest(t, "no blocks", request, fetchRequestNoBlocks)
	})

	t.Run("with properties", func(t *testing.T) {
		request := new(FetchRequest)
		request.MaxWaitTime = 0x20
		request.MinBytes = 0xEF
		testRequest(t, "with properties", request, fetchRequestWithProperties)
	})

	t.Run("one block", func(t *testing.T) {
		request := new(FetchRequest)
		request.MaxWaitTime = 0
		request.MinBytes = 0
		request.AddBlock("topic", 0x12, 0x34, 0x56)
		testRequest(t, "one block", request, fetchRequestOneBlock)
	})

	t.Run("one block v4", func(t *testing.T) {
		request := new(FetchRequest)
		request.Version = 4
		request.MaxBytes = 0xFF
		request.Isolation = ReadCommitted
		request.AddBlock("topic", 0x12, 0x34, 0x56)
		testRequest(t, "one block v4", request, fetchRequestOneBlockV4)
	})

	t.Run("one block v11 rackid", func(t *testing.T) {
		request := new(FetchRequest)
		request.Version = 11
		request.MaxBytes = 0xFF
		request.Isolation = ReadCommitted
		request.SessionID = 0xAA
		request.SessionEpoch = 0xEE
		request.AddBlock("topic", 0x12, 0x34, 0x56)
		request.RackID = "rack01"
		testRequest(t, "one block v11 rackid", request, fetchRequestOneBlockV11)
	})
}