Function TestConsumerOffsetNewest

consumer_test.go:241–283

If `OffsetNewest` is passed as the initial offset then the first consumed message indeed corresponds to the offset that broker claims to be the newest in its metadata response.

(t *testing.T)

// message indeed corresponds to the offset that broker claims to be the
// newest in its metadata response.
func TestConsumerOffsetNewest(t *testing.T) {
	// Given
	offsetNewest := int64(10)
	offsetNewestAfterFetchRequest := int64(50)
	broker0 := NewMockBroker(t, 0)
	broker0.SetHandlerByMap(map[string]MockResponse{
		"MetadataRequest": NewMockMetadataResponse(t).
			SetBroker(broker0.Addr(), broker0.BrokerID()).
			SetLeader("my_topic", 0, broker0.BrokerID()),
		"OffsetRequest": NewMockOffsetResponse(t).
			SetOffset("my_topic", 0, OffsetNewest, offsetNewest).
			SetOffset("my_topic", 0, OffsetOldest, 7),
		"FetchRequest": NewMockFetchResponse(t, 1).
			SetMessage("my_topic", 0, 9, testMsg). // skipped because parseRecords(): offset < child.offset
			SetMessage("my_topic", 0, 10, testMsg).
			SetMessage("my_topic", 0, 11, testMsg).
			SetHighWaterMark("my_topic", 0, offsetNewestAfterFetchRequest),
	})

	master, err := NewConsumer([]string{broker0.Addr()}, NewTestConfig())
	if err != nil {
		t.Fatal(err)
	}

	// When
	consumer, err := master.ConsumePartition("my_topic", 0, OffsetNewest)
	if err != nil {
		t.Fatal(err)
	}

	// Then
	if hwmo := consumer.HighWaterMarkOffset(); hwmo != offsetNewest {
		t.Errorf("Expected high water mark offset %d, found %d", offsetNewest, hwmo)
	}
	assertMessageOffset(t, <-consumer.Messages(), 10)
	if hwmo := consumer.HighWaterMarkOffset(); hwmo != offsetNewestAfterFetchRequest {
		t.Errorf("Expected high water mark offset %d, found %d", offsetNewestAfterFetchRequest, hwmo)
	}

	safeClose(t, consumer)
	safeClose(t, master)
	broker0.Close()
}

// If `OffsetOldest` is passed as the initial offset then the first consumed
// message is indeed the first available in the partition.