Function runTestFuncConsumerGroupMemberWithConfig

functional_consumer_group_test.go:428–459

(
	t *testing.T,
	config *Config,
	groupID string,
	maxMessages int32,
	sink *testFuncConsumerGroupSink,
	topics ...string,
)

}

func runTestFuncConsumerGroupMemberWithConfig(
	t *testing.T,
	config *Config,
	groupID string,
	maxMessages int32,
	sink *testFuncConsumerGroupSink,
	topics ...string,
) *testFuncConsumerGroupMember {
	t.Helper()

	group, err := NewConsumerGroup(FunctionalTestEnv.KafkaBrokerAddrs, groupID, config)
	if err != nil {
		t.Fatal(err)
		return nil
	}

	if len(topics) == 0 {
		topics = []string{"test.4"}
	}

	member := &testFuncConsumerGroupMember{
		ConsumerGroup: group,
		clientID:      config.ClientID,
		claims:        make(map[string]int),
		isCapped:      maxMessages != 0,
		sink:          sink,
		t:             t,
	}
	member.maxMessages.Store(maxMessages)
	go member.loop(topics)
	return member
}

func (m *testFuncConsumerGroupMember) AssertCleanShutdown() {
	m.t.Helper()