func (c *Consumer) release() (err error) {
	// Stop all consumers
	c.subs.Stop()

	// Clear subscriptions on exit
	defer c.subs.Clear()

	// Wait for messages to be processed
	timeout := time.NewTimer(c.client.config.Group.Offsets.Synchronization.DwellTime)
	defer timeout.Stop()

	select {
	case <-c.dying:
	case <-timeout.C:
	}

	// Commit offsets, continue on errors
	if e := c.commitOffsetsWithRetry(c.client.config.Group.Offsets.Retry.Max); e != nil {
		err = e
	}

	return
}
func (c *Consumer) release() (err error) {
	
	c.subs.Stop()

	
	defer c.subs.Clear()

	
	timeout := time.NewTimer(c.client.config.Group.Offsets.Synchronization.DwellTime)
	defer timeout.Stop()

	select {
	case <-c.dying:
	case <-timeout.C:
	}

	
	if e := c.commitOffsetsWithRetry(c.client.config.Group.Offsets.Retry.Max); e != nil {
		err = e
	}

	return
}
func ( c * Consumer ) release ( ) ( err error ) { // Stop all consumers c . subs . Stop ( ) // Clear subscriptions on exit defer c . subs . Clear ( ) // Wait for messages to be processed timeout := time . NewTimer ( c . client . config . Group . Offsets . Synchronization . DwellTime ) defer timeout . Stop ( ) select { case <- c . dying : case <- timeout . C : } // Commit offsets, continue on errors if e := c . commitOffsetsWithRetry ( c . client . config . Group . Offsets . Retry . Max ) ; e != nil { err = e } return }
// Releases the consumer and commits offsets, called from rebalance() and Close()
Releases the consumer and commits offsets called from rebalance () and Close ()