Function TestPreferredReplicaFetchResponse

fetch_response_test.go:584–727

(t *testing.T)

}

func TestPreferredReplicaFetchResponse(t *testing.T) {
	t.Run("decodes v11", func(t *testing.T) {
		response := FetchResponse{}
		testVersionDecodable(
			t, "preferred replica fetch response v11", &response,
			preferredReplicaFetchResponseV11, 11)

		if response.ErrorCode != 0x0002 {
			t.Fatal("Decoding produced incorrect error code.")
		}

		if response.SessionID != 0x000000AC {
			t.Fatal("Decoding produced incorrect session ID.")
		}

		if len(response.Blocks) != 1 {
			t.Fatal("Decoding produced incorrect number of topic blocks.")
		}

		if len(response.Blocks["topic"]) != 1 {
			t.Fatal("Decoding produced incorrect number of partition blocks for topic.")
		}

		block := response.GetBlock("topic", 5)
		if block == nil {
			t.Fatal("GetBlock didn't return block.")
		}
		if !errors.Is(block.Err, ErrOffsetOutOfRange) {
			t.Error("Decoding didn't produce correct error code.")
		}
		if block.HighWaterMarkOffset != 0x10101010 {
			t.Error("Decoding didn't produce correct high water mark offset.")
		}
		if block.LastStableOffset != 0x10101009 {
			t.Error("Decoding didn't produce correct last stable offset.")
		}
		if block.LogStartOffset != 0x01010101 {
			t.Error("Decoding didn't produce correct log start offset.")
		}
		if block.PreferredReadReplica != 0x0003 {
			t.Error("Decoding didn't produce correct preferred read replica.")
		}
		partial, err := block.isPartial()
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		if partial {
			t.Error("Decoding detected a partial trailing record where there wasn't one.")
		}

		n, err := block.numRecords()
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		if n != 1 {
			t.Fatal("Decoding produced incorrect number of records.")
		}
		msgBlock := block.RecordsSet[0].MsgSet.Messages[0]