Function TestListConsumerGroupOffsetsBatch

admin_test.go:1953–2126

(t *testing.T)

}

func TestListConsumerGroupOffsetsBatch(t *testing.T) {
	const (
		topic           = "my-topic"
		groupA          = "group-a"
		groupB          = "group-b"
		expectedOffsetA = int64(7)
		expectedOffsetB = int64(13)
	)
	bothGroups := map[string]map[string][]int32{
		groupA: {topic: {0}},
		groupB: {topic: {0}},
	}

	// setup wires a single mock broker as the coordinator for every named group
	// and serves the given OffsetFetchRequest handler. Returns a v3.0 admin.
	setup := func(t *testing.T, offsetFetch MockResponse, groups ...string) ClusterAdmin {
		t.Helper()
		broker := newMockBroker(t, 1)
		broker.SetHandlerByMap(map[string]MockResponse{
			"OffsetFetchRequest":     offsetFetch,
			"MetadataRequest":        mockMetadataFor(t, broker),
			"FindCoordinatorRequest": mockGroupCoordinators(t, broker, groups...),
		})
		return newTestAdminAt(t, V3_0_0_0, broker)
	}

	// groupBlock builds a v8 response group entry with a single block.
	groupBlock := func(group string, offset int64) OffsetFetchResponseGroup {
		return OffsetFetchResponseGroup{
			GroupId: group,
			Blocks:  map[string]map[int32]*OffsetFetchResponseBlock{topic: {0: {Offset: offset}}},
		}
	}

	t.Run("fetches offsets for multiple groups", func(t *testing.T) {
		admin := setup(t, NewMockOffsetFetchResponse(t).
			SetOffset(groupA, topic, 0, expectedOffsetA, "", ErrNoError).
			SetOffset(groupB, topic, 0, expectedOffsetB, "", ErrNoError).
			SetError(ErrNoError), groupA, groupB)

		result, err := admin.ListConsumerGroupOffsetsBatch(bothGroups)
		require.NoError(t, err)
		assertGroupOffset(t, result, groupA, topic, 0, expectedOffsetA)
		assertGroupOffset(t, result, groupB, topic, 0, expectedOffsetB)
	})

	t.Run("nil partitions fetches all topics for the group", func(t *testing.T) {
		const otherTopic = "other-topic"
		admin := setup(t, NewMockOffsetFetchResponse(t).
			SetOffset(groupA, topic, 0, expectedOffsetA, "", ErrNoError).
			SetOffset(groupA, otherTopic, 0, expectedOffsetB, "", ErrNoError).
			SetError(ErrNoError), groupA)

		result, err := admin.ListConsumerGroupOffsetsBatch(map[string]map[string][]int32{groupA: nil})
		require.NoError(t, err)
		assertGroupOffset(t, result, groupA, topic, 0, expectedOffsetA)
		assertGroupOffset(t, result, groupA, otherTopic, 0, expectedOffsetB)
	})