Function TestConsumerOffsetsAreManagedCorrectlyWithOffsetOldest

mocks/consumer_test.go:337–367

(t *testing.T)

}

func TestConsumerOffsetsAreManagedCorrectlyWithOffsetOldest(t *testing.T) {
	trm := newTestReporterMock()
	consumer := NewConsumer(trm, NewTestConfig())
	pcmock := consumer.ExpectConsumePartition("test", 0, sarama.OffsetOldest)
	pcmock.YieldMessage(&sarama.ConsumerMessage{Value: []byte("hello")})
	pcmock.YieldMessage(&sarama.ConsumerMessage{Value: []byte("hello")})
	pcmock.ExpectMessagesDrainedOnClose()

	pc, err := consumer.ConsumePartition("test", 0, sarama.OffsetOldest)
	if err != nil {
		t.Error(err)
	}

	message1 := <-pc.Messages()
	if message1.Offset != 0 {
		t.Errorf("Expected offset of first message in the partition to be 0, got %d", message1.Offset)
	}

	message2 := <-pc.Messages()
	if message2.Offset != 1 {
		t.Errorf("Expected offset of second message in the partition to be 1, got %d", message2.Offset)
	}

	if err := consumer.Close(); err != nil {
		t.Error(err)
	}

	if len(trm.errors) != 0 {
		t.Errorf("Expected to not report any errors, found: %v", trm.errors)
	}
}

func TestConsumerOffsetsAreManagedCorrectlyWithSpecifiedOffset(t *testing.T) {
	startingOffset := int64(123)