Method Start

consumergroup.go:375–413

Start launches the provided function in a go routine and adds accounting such that when the function exits, it stops the current generation (if not already in the process of doing so). The provided function MUST support cancellation via the ctx argument and exit in a timely manner once the ctx is c

(fn func(ctx context.Context))

// progress for this consumer and potentially cause consumer group membership
// churn.
func (g *Generation) Start(fn func(ctx context.Context)) {
	g.lock.Lock()
	defer g.lock.Unlock()

	// this is an edge case: if the generation has already closed, then it's
	// possible that the close func has already waited on outstanding go
	// routines and exited.
	//
	// nonetheless, it's important to honor that the fn is invoked in case the
	// calling function is waiting e.g. on a channel send or a WaitGroup.  in
	// such a case, fn should immediately exit because ctx.Err() will return
	// ErrGenerationEnded.
	if g.closed {
		go fn(genCtx{g})
		return
	}

	// register that there is one more go routine that's part of this gen.
	g.routines++

	go func() {
		fn(genCtx{g})
		g.lock.Lock()
		// shut down the generation as soon as one function exits.  this is
		// different from close() in that it doesn't wait for all go routines in
		// the generation to exit.
		if !g.closed {
			close(g.done)
			g.closed = true
		}
		g.routines--
		// if this was the last go routine in the generation, close the joined
		// chan so that close() can exit if it's waiting.
		if g.routines == 0 {
			close(g.joined)
		}
		g.lock.Unlock()
	}()
}

// CommitOffsets commits the provided topic+partition+offset combos to the
// consumer group coordinator.  This can be used to reset the consumer to