Method UpdateMembership

dgraph/cmd/zero/zero.go:732–786

UpdateMembership updates the membership of the given group.

(ctx context.Context, group *pb.Group)

// UpdateMembership updates the membership of the given group.
func (s *Server) UpdateMembership(ctx context.Context, group *pb.Group) (*api.Payload, error) {
	// Only Zero leader would get these membership updates.
	if ts := group.GetCheckpointTs(); ts > 0 {
		for _, m := range group.GetMembers() {
			s.Lock()
			s.checkpointPerGroup[m.GetGroupId()] = ts
			s.Unlock()
		}
	}
	proposals, err := s.createProposals(group)
	if err != nil {
		// Sleep here so the caller doesn't keep on retrying indefinitely, creating a busy
		// wait.
		time.Sleep(time.Second)
		glog.Errorf("Error while creating proposals in Update: %v\n", err)
		return nil, err
	}

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	errCh := make(chan error, len(proposals))
	for _, pr := range proposals {
		go func(pr *pb.ZeroProposal) {
			errCh <- s.Node.proposeAndWait(ctx, pr)
		}(pr)
	}

	for range proposals {
		// We Don't care about these errors
		// Ideally shouldn't error out.
		if err := <-errCh; err != nil {
			glog.Errorf("Error while applying proposal in Update stream: %v\n", err)
			return nil, err
		}
	}

	if len(group.Members) == 0 {
		return &api.Payload{Data: []byte("OK")}, nil
	}
	select {
	case s.moveOngoing <- struct{}{}:
	default:
		// If a move is going on, don't do the next steps of deleting predicates.
		return &api.Payload{Data: []byte("OK")}, nil
	}
	defer func() {
		<-s.moveOngoing
	}()

	if err := s.deletePredicates(ctx, group); err != nil {
		glog.Warningf("While deleting predicates: %v", err)
	}
	return &api.Payload{Data: []byte("OK")}, nil
}

func (s *Server) deletePredicates(ctx context.Context, group *pb.Group) error {
	if group == nil || group.Tablets == nil {