def _on_join_leader(self, response):
        """
        Perform leader synchronization and send back the assignment
        for the group via SyncGroupRequest

        Arguments:
            response (JoinResponse): broker response to parse

        Returns:
            Future: resolves to member assignment encoded-bytes
        """
        try:
            group_assignment = self._perform_assignment(response.leader_id,
                                                        response.group_protocol,
                                                        response.members)
        except Exception as e:
            return Future().failure(e)

        version = self._client.api_version(SyncGroupRequest, max_version=2)
        request = SyncGroupRequest[version](
            self.group_id,
            self._generation.generation_id,
            self._generation.member_id,
            [(member_id,
              assignment if isinstance(assignment, bytes) else assignment.encode())
             for member_id, assignment in six.iteritems(group_assignment)])

        log.debug("Sending leader SyncGroup for group %s to coordinator %s: %s",
                  self.group_id, self.coordinator_id, request)
        return self._send_sync_group_request(request)

        Perform leader synchronization and send back the assignment
        for the group via SyncGroupRequest

        Arguments:
            response (JoinResponse): broker response to parse

        Returns:
            Future: resolves to member assignment encoded-bytes
        
kafka/coordinator/base.py