Function TestAlterConsumerGroupOffsets

admin_test.go:2303–2352

(t *testing.T)

}

func TestAlterConsumerGroupOffsets(t *testing.T) {
	const (
		group     = "my-group"
		topic     = "my-topic"
		partition = int32(0)
	)
	offsets := map[string]map[int32]OffsetAndMetadata{
		topic: {partition: {Offset: 100, LeaderEpoch: -1}},
	}

	t.Run("commits offsets via the group coordinator", func(t *testing.T) {
		broker := newMockBroker(t, 1)
		broker.SetHandlerByMap(map[string]MockResponse{
			"OffsetCommitRequest":    NewMockOffsetCommitResponse(t).SetError(group, topic, partition, ErrNoError),
			"FindCoordinatorRequest": NewMockFindCoordinatorResponse(t).SetCoordinator(CoordinatorGroup, group, broker),
			"MetadataRequest":        mockMetadataFor(t, broker),
		})

		response, err := newTestAdmin(t, broker).AlterConsumerGroupOffsets(group, offsets, nil)
		require.NoError(t, err)
		assert.Equal(t, ErrNoError, response.Errors[topic][partition])
	})

	t.Run("returns ConfigurationError when no offsets provided", func(t *testing.T) {
		broker := newMockBroker(t, 1)
		broker.SetHandlerByMap(map[string]MockResponse{"MetadataRequest": mockMetadataFor(t, broker)})

		response, err := newTestAdmin(t, broker).AlterConsumerGroupOffsets(group, nil, nil)
		assert.Nil(t, response)

		var cfgErr ConfigurationError
		require.ErrorAs(t, err, &cfgErr)
	})

	t.Run("retries on per-partition NOT_COORDINATOR", func(t *testing.T) {
		broker := newMockBroker(t, 1)
		broker.SetHandlerByMap(map[string]MockResponse{
			"OffsetCommitRequest": NewMockSequence(
				NewMockOffsetCommitResponse(t).SetError(group, topic, partition, ErrNotCoordinatorForConsumer),
				NewMockOffsetCommitResponse(t).SetError(group, topic, partition, ErrNoError),
			),
			"FindCoordinatorRequest": NewMockFindCoordinatorResponse(t).SetCoordinator(CoordinatorGroup, group, broker),
			"MetadataRequest":        mockMetadataFor(t, broker),
		})

		response, err := newTestAdmin(t, broker).AlterConsumerGroupOffsets(group, offsets, nil)
		require.NoError(t, err)
		assert.Equal(t, ErrNoError, response.Errors[topic][partition])
	})
}

func TestDeleteConsumerGroup(t *testing.T) {
	seedBroker := NewMockBroker(t, 1)