func (c *Consumer) syncGroup(strategy *balancer) (map[string][]int32, error) {
	memberID, generationID := c.membership()
	req := &sarama.SyncGroupRequest{
		GroupId:      c.groupID,
		MemberId:     memberID,
		GenerationId: generationID,
	}

	if strategy != nil {
		for memberID, topics := range strategy.Perform() {
			if err := req.AddGroupAssignmentMember(memberID, &sarama.ConsumerGroupMemberAssignment{
				Topics: topics,
			}); err != nil {
				return nil, err
			}
		}
	}

	broker, err := c.client.Coordinator(c.groupID)
	if err != nil {
		c.closeCoordinator(broker, err)
		return nil, err
	}

	resp, err := broker.SyncGroup(req)
	if err != nil {
		c.closeCoordinator(broker, err)
		return nil, err
	} else if resp.Err != sarama.ErrNoError {
		c.closeCoordinator(broker, resp.Err)
		return nil, resp.Err
	}

	// Return if there is nothing to subscribe to
	if len(resp.MemberAssignment) == 0 {
		return nil, nil
	}

	// Get assigned subscriptions
	members, err := resp.GetMemberAssignment()
	if err != nil {
		return nil, err
	}

	// Sort partitions, for each topic
	for topic := range members.Topics {
		sort.Sort(int32Slice(members.Topics[topic]))
	}
	return members.Topics, nil
}
func (c *Consumer) syncGroup(strategy *balancer) (map[string][]int32, error) {
	memberID, generationID := c.membership()
	req := &sarama.SyncGroupRequest{
		GroupId:      c.groupID,
		MemberId:     memberID,
		GenerationId: generationID,
	}

	if strategy != nil {
		for memberID, topics := range strategy.Perform() {
			if err := req.AddGroupAssignmentMember(memberID, &sarama.ConsumerGroupMemberAssignment{
				Topics: topics,
			}); err != nil {
				return nil, err
			}
		}
	}

	broker, err := c.client.Coordinator(c.groupID)
	if err != nil {
		c.closeCoordinator(broker, err)
		return nil, err
	}

	resp, err := broker.SyncGroup(req)
	if err != nil {
		c.closeCoordinator(broker, err)
		return nil, err
	} else if resp.Err != sarama.ErrNoError {
		c.closeCoordinator(broker, resp.Err)
		return nil, resp.Err
	}

	
	if len(resp.MemberAssignment) == 0 {
		return nil, nil
	}

	
	members, err := resp.GetMemberAssignment()
	if err != nil {
		return nil, err
	}

	
	for topic := range members.Topics {
		sort.Sort(int32Slice(members.Topics[topic]))
	}
	return members.Topics, nil
}
func ( c * Consumer ) syncGroup ( strategy * balancer ) ( map [ string ] [ ] int32 , error ) { memberID , generationID := c . membership ( ) req := & sarama . SyncGroupRequest { GroupId : c . groupID , MemberId : memberID , GenerationId : generationID , } if strategy != nil { for memberID , topics := range strategy . Perform ( ) { if err := req . AddGroupAssignmentMember ( memberID , & sarama . ConsumerGroupMemberAssignment { Topics : topics , } ) ; err != nil { return nil , err } } } broker , err := c . client . Coordinator ( c . groupID ) if err != nil { c . closeCoordinator ( broker , err ) return nil , err } resp , err := broker . SyncGroup ( req ) if err != nil { c . closeCoordinator ( broker , err ) return nil , err } else if resp . Err != sarama . ErrNoError { c . closeCoordinator ( broker , resp . Err ) return nil , resp . Err } // Return if there is nothing to subscribe to if len ( resp . MemberAssignment ) == 0 { return nil , nil } // Get assigned subscriptions members , err := resp . GetMemberAssignment ( ) if err != nil { return nil , err } // Sort partitions, for each topic for topic := range members . Topics { sort . Sort ( int32Slice ( members . Topics [ topic ] ) ) } return members . Topics , nil }
// Send a request to the broker to sync the group on rebalance().
// Returns a list of topics and partitions to consume.
Send a request to the broker to sync the group on rebalance () . Returns a list of topics and partitions to consume .