Function TestConsumerMessageWithKey

consumer_test.go:93–155

If a message is given a key, it can be correctly collected while consuming.

(t *testing.T)

// If a message is given a key, it can be correctly collected while consuming.
func TestConsumerMessageWithKey(t *testing.T) {
	// Given
	broker0 := NewMockBroker(t, 0)

	manualOffset := int64(1234)
	offsetNewest := int64(2345)
	offsetNewestAfterFetchRequest := int64(3456)

	mockFetchResponse := NewMockFetchResponse(t, 1)

	// skipped because parseRecords(): offset < child.offset
	mockFetchResponse.SetMessageWithKey("my_topic", 0, manualOffset-1, testKey, testMsg)

	for i := range int64(10) {
		mockFetchResponse.SetMessageWithKey("my_topic", 0, i+manualOffset, testKey, testMsg)
	}

	mockFetchResponse.SetHighWaterMark("my_topic", 0, offsetNewestAfterFetchRequest)

	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, OffsetOldest, 0).
			SetOffset("my_topic", 0, OffsetNewest, offsetNewest),
		"FetchRequest": mockFetchResponse,
	})

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

	consumer, err := master.ConsumePartition("my_topic", 0, manualOffset)
	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)
	}
	for i := range int64(10) {
		select {
		case message := <-consumer.Messages():
			assertMessageOffset(t, message, i+manualOffset)
			assertMessageKey(t, message, testKey)
			assertMessageValue(t, message, testMsg)
		case err := <-consumer.Errors():
			t.Error(err)
		}
	}

	if hwmo := consumer.HighWaterMarkOffset(); hwmo != offsetNewestAfterFetchRequest {
		t.Errorf("Expected high water mark offset %d, found %d", offsetNewestAfterFetchRequest, hwmo)
	}