Function newOffsetManagerFromClient

offset_manager.go:57–86

(group, memberID string, generation int32, client Client, sessionCanceler context.CancelCauseFunc)

}

func newOffsetManagerFromClient(group, memberID string, generation int32, client Client, sessionCanceler context.CancelCauseFunc) (*offsetManager, error) {
	// Check that we are not dealing with a closed Client before processing any other arguments
	if client.Closed() {
		return nil, ErrClosedClient
	}

	conf := client.Config()
	om := &offsetManager{
		client:          client,
		conf:            conf,
		group:           group,
		poms:            make(map[string]map[int32]*partitionOffsetManager),
		sessionCanceler: sessionCanceler,

		memberID:   memberID,
		generation: generation,

		closing: make(chan none),
		closed:  make(chan none),
	}
	if conf.Consumer.Group.InstanceId != "" {
		om.groupInstanceId = &conf.Consumer.Group.InstanceId
	}
	if conf.Consumer.Offsets.AutoCommit.Enable {
		om.ticker = time.NewTicker(conf.Consumer.Offsets.AutoCommit.Interval)
		go withRecover(om.mainLoop)
	}

	return om, nil
}

func (om *offsetManager) ManagePartition(topic string, partition int32) (PartitionOffsetManager, error) {
	pom, err := om.newPartitionOffsetManager(topic, partition)