Method joinGroup

consumergroup.go:934–977

joinGroup attempts to join the reader to the consumer group. Returns GroupMemberAssignments is this Reader was selected as the leader. Otherwise, GroupMemberAssignments will be nil. Possible kafka error codes returned: * GroupLoadInProgress: * GroupCoordinatorNotAvailable: * NotCoordinatorForGroup

(conn coordinator, memberID string)

//  * InvalidSessionTimeout:
//  * GroupAuthorizationFailed:
func (cg *ConsumerGroup) joinGroup(conn coordinator, memberID string) (string, int32, GroupMemberAssignments, error) {
	request, err := cg.makeJoinGroupRequest(memberID)
	if err != nil {
		return "", 0, nil, err
	}

	response, err := conn.joinGroup(request)
	if err == nil && response.ErrorCode != 0 {
		err = Error(response.ErrorCode)
	}
	if err != nil {
		return "", 0, nil, err
	}

	memberID = response.MemberID
	generationID := response.GenerationID

	cg.withLogger(func(l Logger) {
		l.Printf("joined group %s as member %s in generation %d", cg.config.ID, memberID, generationID)
	})

	var assignments GroupMemberAssignments
	if iAmLeader := response.MemberID == response.LeaderID; iAmLeader {
		v, err := cg.assignTopicPartitions(conn, response)
		if err != nil {
			return memberID, 0, nil, err
		}
		assignments = v

		cg.withLogger(func(l Logger) {
			for memberID, assignment := range assignments {
				for topic, partitions := range assignment {
					l.Printf("assigned member/topic/partitions %v/%v/%v", memberID, topic, partitions)
				}
			}
		})
	}

	cg.withLogger(func(l Logger) {
		l.Printf("joinGroup succeeded for response, %v.  generationID=%v, memberID=%v", cg.config.ID, response.GenerationID, response.MemberID)
	})

	return memberID, generationID, assignments, nil
}

// makeJoinGroupRequestV1 handles the logic of constructing a joinGroup
// request.