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)

	// 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()