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)
		request.ReplicaID = -1
		request.Topics = []FetchableTopic{}
		testRequest(t, "no blocks", request, fetchRequestNoBlocks)
	})

	t.Run("with properties", func(t *testing.T) {
		request := new(FetchRequest)
		request.ReplicaID = -1
		request.MaxWait = 0x20
		request.MinBytes = 0xEF
		request.Topics = []FetchableTopic{}
		testRequest(t, "with properties", request, fetchRequestWithProperties)
	})

	// AddBlock(topic string, partitionID int32, fetchOffset int64, maxBytes int32) {
	t.Run("one block", func(t *testing.T) {
		request := new(FetchRequest)
		request.ReplicaID = -1
		request.MaxWait = 0
		request.MinBytes = 0
		request.Topics = []FetchableTopic{
			{
				Name: "topic",
				FetchPartitions: []FetchPartition{
					{
						PartitionIndex: 0x12,
						FetchOffset:    0x34,
						MaxBytes:       0x56,
					},
				},
			},
		}
		testRequest(t, "one block", request, fetchRequestOneBlock)
	})

	t.Run("one block v4", func(t *testing.T) {
		request := new(FetchRequest)
		request.Version = 4
		request.ReplicaID = -1
		request.MaxBytes = 0xFF
		request.IsolationLevel = ReadCommitted
		request.Topics = []FetchableTopic{
			{
				Version: 4,
				Name:    "topic",
				FetchPartitions: []FetchPartition{
					{
						Version:        4,
						PartitionIndex: 0x12,
						FetchOffset:    0x34,
						MaxBytes:       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.ReplicaID = -1
		request.MaxBytes = 0xFF
		request.IsolationLevel = ReadCommitted
		request.SessionID = 0xAA
		request.Epoch = 0xEE
		request.Topics = []FetchableTopic{
			{
				Version: 11,
				Name:    "topic",
				FetchPartitions: []FetchPartition{
					{
						Version:            11,
						PartitionIndex:     0x12,
						CurrentLeaderEpoch: -1,
						FetchOffset:        0x34,
						MaxBytes:           0x56,
					},
				},
			},
		}
		request.Forgotten = []ForgottenTopic{}
		request.RackID = "rack01"
		testRequest(t, "one block v11 rackid", request, fetchRequestOneBlockV11)
	})
}