func (c *Consumer) rebalance() (map[string][]int32, error) {
	memberID, _ := c.membership()
	sarama.Logger.Printf("cluster/consumer %s rebalance\n", memberID)

	allTopics, err := c.client.Topics()
	if err != nil {
		return nil, err
	}
	c.extraTopics = c.selectExtraTopics(allTopics)
	sort.Strings(c.extraTopics)

	// Re-join consumer group
	strategy, err := c.joinGroup()
	switch {
	case err == sarama.ErrUnknownMemberId:
		c.membershipMu.Lock()
		c.memberID = ""
		c.membershipMu.Unlock()
		return nil, err
	case err != nil:
		return nil, err
	}

	// Sync consumer group state, fetch subscriptions
	subs, err := c.syncGroup(strategy)
	switch {
	case err == sarama.ErrRebalanceInProgress:
		return nil, err
	case err != nil:
		_ = c.leaveGroup()
		return nil, err
	}
	return subs, nil
}
func (c *Consumer) rebalance() (map[string][]int32, error) {
	memberID, _ := c.membership()
	sarama.Logger.Printf("cluster/consumer %s rebalance\n", memberID)

	allTopics, err := c.client.Topics()
	if err != nil {
		return nil, err
	}
	c.extraTopics = c.selectExtraTopics(allTopics)
	sort.Strings(c.extraTopics)

	
	strategy, err := c.joinGroup()
	switch {
	case err == sarama.ErrUnknownMemberId:
		c.membershipMu.Lock()
		c.memberID = ""
		c.membershipMu.Unlock()
		return nil, err
	case err != nil:
		return nil, err
	}

	
	subs, err := c.syncGroup(strategy)
	switch {
	case err == sarama.ErrRebalanceInProgress:
		return nil, err
	case err != nil:
		_ = c.leaveGroup()
		return nil, err
	}
	return subs, nil
}
func ( c * Consumer ) rebalance ( ) ( map [ string ] [ ] int32 , error ) { memberID , _ := c . membership ( ) sarama . Logger . Printf ( " \n " , memberID ) allTopics , err := c . client . Topics ( ) if err != nil { return nil , err } c . extraTopics = c . selectExtraTopics ( allTopics ) sort . Strings ( c . extraTopics ) // Re-join consumer group strategy , err := c . joinGroup ( ) switch { case err == sarama . ErrUnknownMemberId : c . membershipMu . Lock ( ) c . memberID = " " c . membershipMu . Unlock ( ) return nil , err case err != nil : return nil , err } // Sync consumer group state, fetch subscriptions subs , err := c . syncGroup ( strategy ) switch { case err == sarama . ErrRebalanceInProgress : return nil , err case err != nil : _ = c . leaveGroup ( ) return nil , err } return subs , nil }
// Performs a rebalance, part of the mainLoop()
Performs a rebalance part of the mainLoop ()