Function TestConsumeMessagesTrackLeader

consumer_test.go:1055–1164

TestConsumeMessagesTrackLeader ensures that in the event that leadership of a topicPartition changes and no preferredReadReplica is specified, the consumer connects back to the new leader to resume consumption and doesn't continue consuming from the follower. See https://github.com/IBM/sarama/issue

(t *testing.T)

//
// See https://github.com/IBM/sarama/issues/1927
func TestConsumeMessagesTrackLeader(t *testing.T) {
	cfg := NewTestConfig()
	cfg.ClientID = t.Name()
	cfg.Metadata.RefreshFrequency = time.Millisecond * 50
	cfg.Consumer.Retry.Backoff = 0
	cfg.Net.MaxOpenRequests = 1
	cfg.Version = V2_1_0_0

	leader1 := NewMockBroker(t, 1)
	leader2 := NewMockBroker(t, 2)

	mockMetadataResponse1 := NewMockMetadataResponse(t).
		SetBroker(leader1.Addr(), leader1.BrokerID()).
		SetBroker(leader2.Addr(), leader2.BrokerID()).
		SetLeader("my_topic", 0, leader1.BrokerID())
	mockMetadataResponse2 := NewMockMetadataResponse(t).
		SetBroker(leader1.Addr(), leader1.BrokerID()).
		SetBroker(leader2.Addr(), leader2.BrokerID()).
		SetLeader("my_topic", 0, leader2.BrokerID())
	mockMetadataResponse3 := NewMockMetadataResponse(t).
		SetBroker(leader1.Addr(), leader1.BrokerID()).
		SetBroker(leader2.Addr(), leader2.BrokerID()).
		SetLeader("my_topic", 0, leader1.BrokerID())

	leader1.SetHandlerByMap(map[string]MockResponse{
		"MetadataRequest": mockMetadataResponse1,
		"OffsetRequest": NewMockOffsetResponse(t).
			SetOffset("my_topic", 0, OffsetNewest, 1234).
			SetOffset("my_topic", 0, OffsetOldest, 0),
		"FetchRequest": NewMockFetchResponse(t, 1).
			SetMessage("my_topic", 0, 1, testMsg).
			SetMessage("my_topic", 0, 2, testMsg),
	})

	leader2.SetHandlerByMap(map[string]MockResponse{
		"MetadataRequest": mockMetadataResponse1,
	})

	client, err := NewClient([]string{leader1.Addr()}, cfg)
	if err != nil {
		t.Fatal(err)
	}

	consumer, err := NewConsumerFromClient(client)
	if err != nil {
		t.Fatal(err)
	}

	pConsumer, err := consumer.ConsumePartition("my_topic", 0, 1)
	if err != nil {
		t.Fatal(err)
	}

	assertMessageOffset(t, <-pConsumer.Messages(), 1)
	assertMessageOffset(t, <-pConsumer.Messages(), 2)

	fetchEmptyResponse := &FetchResponse{Version: 10}
	fetchEmptyResponse.AddError("my_topic", 0, ErrNoError)