pass through protocol name for sync group

Author: rhansen2

consumergroup.go

@@ -812,11 +812,12 @@ func (cg *ConsumerGroup) nextGeneration(memberID string) (string, error) {
 	var generationID int
 	var groupAssignments GroupMemberAssignments
 	var assignments map[string][]int
+	var protocolName string
 	var err error
 
 	// join group.  this will join the group and prepare assignments if our
 	// consumer is elected leader.  it may also change or assign the member ID.
-	memberID, generationID, groupAssignments, err = cg.joinGroup(memberID)
+	memberID, generationID, protocolName, groupAssignments, err = cg.joinGroup(memberID)
 	if err != nil {
 		cg.withErrorLogger(func(log Logger) {
 			log.Printf("Failed to join group %s: %v", cg.config.ID, err)
@@ -828,7 +829,7 @@ func (cg *ConsumerGroup) nextGeneration(memberID string) (string, error) {
 	})
 
 	// sync group
-	assignments, err = cg.syncGroup(memberID, generationID, groupAssignments)
+	assignments, err = cg.syncGroup(memberID, generationID, protocolName, groupAssignments)
 	if err != nil {
 		cg.withErrorLogger(func(log Logger) {
 			log.Printf("Failed to sync group %s: %v", cg.config.ID, err)
@@ -905,10 +906,10 @@ func (cg *ConsumerGroup) nextGeneration(memberID string) (string, error) {
 //  * InconsistentGroupProtocol:
 //  * InvalidSessionTimeout:
 //  * GroupAuthorizationFailed:
-func (cg *ConsumerGroup) joinGroup(memberID string) (string, int, GroupMemberAssignments, error) {
+func (cg *ConsumerGroup) joinGroup(memberID string) (string, int, string, GroupMemberAssignments, error) {
 	request, err := cg.makeJoinGroupRequest(memberID)
 	if err != nil {
-		return "", 0, nil, err
+		return "", 0, "", nil, err
 	}
 
 	response, err := cg.coord.joinGroup(request)
@@ -919,20 +920,19 @@ func (cg *ConsumerGroup) joinGroup(memberID string) (string, int, GroupMemberAss
 		memberID = response.MemberID
 	}
 	if err != nil {
-		return memberID, 0, nil, err
+		return memberID, 0, "", nil, err
 	}
 
 	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(response)
 		if err != nil {
-			return memberID, 0, nil, err
+			return memberID, 0, "", nil, err
 		}
 		assignments = v
 
@@ -949,7 +949,7 @@ func (cg *ConsumerGroup) joinGroup(memberID string) (string, int, GroupMemberAss
 		l.Printf("joinGroup succeeded for response, %v.  generationID=%v, memberID=%v", cg.config.ID, response.GenerationID, response.MemberID)
 	})
 
-	return memberID, generationID, assignments, nil
+	return memberID, generationID, response.ProtocolName, assignments, nil
 }
 
 // makeJoinGroupRequest handles the logic of constructing a joinGroup
@@ -1044,8 +1044,8 @@ func (cg *ConsumerGroup) assignTopicPartitions(group *JoinGroupResponse) (GroupM
 //  * IllegalGeneration:
 //  * RebalanceInProgress:
 //  * GroupAuthorizationFailed:
-func (cg *ConsumerGroup) syncGroup(memberID string, generationID int, memberAssignments GroupMemberAssignments) (map[string][]int, error) {
-	request := cg.makeSyncGroupRequest(memberID, generationID, memberAssignments)
+func (cg *ConsumerGroup) syncGroup(memberID string, generationID int, protocolName string, memberAssignments GroupMemberAssignments) (map[string][]int, error) {
+	request := cg.makeSyncGroupRequest(memberID, generationID, protocolName, memberAssignments)
 	response, err := cg.coord.syncGroup(request)
 	if err == nil && response.Error != nil {
 		err = response.Error
@@ -1066,11 +1066,13 @@ func (cg *ConsumerGroup) syncGroup(memberID string, generationID int, memberAssi
 	return response.Assignment.AssignedPartitions, nil
 }
 
-func (cg *ConsumerGroup) makeSyncGroupRequest(memberID string, generationID int, memberAssignments GroupMemberAssignments) *SyncGroupRequest {
+func (cg *ConsumerGroup) makeSyncGroupRequest(memberID string, generationID int, protocolName string, memberAssignments GroupMemberAssignments) *SyncGroupRequest {
 	request := &SyncGroupRequest{
 		GroupID:      cg.config.ID,
 		GenerationID: generationID,
 		MemberID:     memberID,
+		ProtocolType: defaultProtocolType,
+		ProtocolName: protocolName,
 	}
 
 	if memberAssignments != nil {