Method RemoveMemberFromConsumerGroup

admin.go:1723–1762

(group string, groupInstanceIds []string)

}

func (ca *clusterAdmin) RemoveMemberFromConsumerGroup(group string, groupInstanceIds []string) (*LeaveGroupResponse, error) {
	if !ca.conf.Version.IsAtLeast(V2_4_0_0) {
		return nil, ConfigurationError("Removing members from a consumer group headers requires Kafka version of at least v2.4.0")
	}
	var response *LeaveGroupResponse
	request := &LeaveGroupRequest{
		Version: 3,
		GroupId: group,
	}
	for _, instanceId := range groupInstanceIds {
		groupInstanceId := instanceId
		request.Members = append(request.Members, MemberIdentity{
			GroupInstanceId: &groupInstanceId,
		})
	}
	err := ca.retryOnError(isRetriableGroupCoordinatorError, func() (err error) {
		defer func() {
			if err != nil && isRetriableGroupCoordinatorError(err) {
				_ = ca.client.RefreshCoordinator(group)
			}
		}()

		coordinator, err := ca.client.Coordinator(group)
		if err != nil {
			return err
		}

		response, err = coordinator.LeaveGroup(request)
		if err != nil {
			return err
		}
		if !errors.Is(response.Err, ErrNoError) {
			return response.Err
		}

		return nil
	})

	return response, err
}