Function TestConsumeMessagesFromReadReplica

consumer_test.go:845–950

(t *testing.T)

}

func TestConsumeMessagesFromReadReplica(t *testing.T) {
	withRefreshFrequency := func(frequency time.Duration) func(*Config) {
		return func(cfg *Config) {
			cfg.Metadata.RefreshFrequency = frequency
		}
	}
	preferredReplica := func(replicaID int32) preferredReadReplica {
		return preferredReadReplica{id: replicaID, ok: true}
	}
	assertOffsets := func(t *testing.T, c PartitionConsumer, offsets ...int64) {
		t.Helper()
		for _, want := range offsets {
			select {
			case msg := <-c.Messages():
				assertMessageOffset(t, msg, want)
			case <-time.After(5 * time.Second):
				require.Failf(t, "timed out waiting for message", "offset %d", want)
			}
		}
	}

	t.Run("switches to preferred follower", func(t *testing.T) {
		c, cleanup := newReadReplicaTest(t, readReplicaTestConfig{
			leaderFetches: []readReplicaFetch{
				{records: []int64{1, 2}, preferredReadReplica: preferredReplica(1)},
			},
			followerFetches: []readReplicaFetch{
				{records: []int64{3, 4}},
			},
		})
		defer cleanup()
		assertOffsets(t, c, 1, 2, 3, 4)
	})

	t.Run("falls back to leader when preferred replica is unknown", func(t *testing.T) {
		const notInMetadata = 5
		c, cleanup := newReadReplicaTest(t, readReplicaTestConfig{
			leaderFetches: []readReplicaFetch{
				{records: []int64{1, 2}, preferredReadReplica: preferredReplica(notInMetadata)},
				{records: []int64{3, 4}},
			},
		})
		defer cleanup()
		assertOffsets(t, c, 1, 2, 3, 4)
	})

	t.Run("falls back to leader on ErrReplicaNotAvailable", func(t *testing.T) {
		c, cleanup := newReadReplicaTest(t, readReplicaTestConfig{
			leaderFetches: []readReplicaFetch{
				{preferredReadReplica: preferredReplica(1)},
				{records: []int64{3, 4}},
			},
			followerFetches: []readReplicaFetch{
				{records: []int64{1, 2}},
				{err: ErrReplicaNotAvailable},
			},
		})
		defer cleanup()