TODO: This method needs a leader session ID
6458
@Override
	public void declineCheckpoint(DeclineCheckpoint decline) {
		final CheckpointCoordinator checkpointCoordinator = executionGraph.getCheckpointCoordinator();

		if (checkpointCoordinator != null) {
			getRpcService().execute(() -> {
				try {
					checkpointCoordinator.receiveDeclineMessage(decline);
				} catch (Exception e) {
					log.error("Error in CheckpointCoordinator while processing {}", decline, e);
				}
			});
		} else {
			String errorMessage = "Received DeclineCheckpoint message for job {} with no CheckpointCoordinator";
			if (executionGraph.getState() == JobStatus.RUNNING) {
				log.error(errorMessage, jobGraph.getJobID());
			} else {
				log.debug(errorMessage, jobGraph.getJobID());
			}
		}
	}
164555
@Override
	public void notifyLeaderAddress(String leaderAddress, UUID leaderSessionID) {
		future.complete(new LeaderAddressAndId(leaderAddress, leaderSessionID));
	}
336774
public PMSession registerSession(String sessionId) {
        synchronized (sessions) {
            if (sessionId != null) {
                if (!sessions.containsKey(sessionId)) {
                    sessions.put(sessionId, new PMSession(sessionId));
                }
                return getSession(sessionId);
            } else {
                return registerSession(newSessionId());
            }
        }
    }
860341
@Override
    @Synchronized
    public void takeLeadership(CuratorFramework client) throws Exception {
        log.info("Obtained leadership to monitor the Host to Segment Container Mapping");

        //Attempt a rebalance whenever leadership is obtained to ensure no host events are missed.
        hostsChange.release();

        //Start cluster monitor.
        pravegaServiceCluster = new ClusterZKImpl(client, ClusterType.HOST);

        //Add listener to track host changes on the monitored pravega cluster.
        pravegaServiceCluster.addListener((type, host) -> {
            switch (type) {
                case HOST_ADDED:
                case HOST_REMOVED:
                    //We don't keep track of the hosts and we always query for the entire set from the cluster
                    //when changes occur. This is to avoid any inconsistencies if we miss any notifications.
                    log.info("Received event: {} for host: {}. Wake up leader for rebalancing", type, host);
                    hostsChange.release();
                    break;
                case ERROR:
                    //This event should be due to ZK connection errors and would have been received by the monitor too,
                    //hence not handling it explicitly here.
                    log.info("Received error event when monitoring the pravega host cluster, ignoring...");
                    break;
            }
        });

        //Keep looping here as long as possible to stay as the leader and exclusively monitor the pravega cluster.
        while (true) {
            try {
                if (suspended.get()) {
                    log.info("Monitor is suspended, waiting for notification to resume");
                    suspendMonitor.acquire();
                    log.info("Resuming monitor");
                }

                hostsChange.acquire();
                log.info("Received rebalance event");

                // Wait here until rebalance can be performed.
                waitForRebalance();

                // Clear all events that has been received until this point since this will be included in the current
                // rebalance operation.
                hostsChange.drainPermits();
                triggerRebalance();
            } catch (InterruptedException e) {
                log.warn("Leadership interrupted, releasing monitor thread");

                //Stop watching the pravega cluster.
                pravegaServiceCluster.close();
                throw e;
            } catch (Exception e) {
                //We will not release leadership if in suspended mode.
                if (!suspended.get()) {
                    log.warn("Failed to perform rebalancing, relinquishing leadership");

                    //Stop watching the pravega cluster.
                    pravegaServiceCluster.close();
                    throw e;
                }
            }
        }
    }
1240049
public final long getSession(RaftGroupId groupId) {
        SessionState session = sessions.get(groupId);
        return session != null ? session.id : NO_SESSION_ID;
    }
1317126
private void addChild(Session session, GroupImpl parent, GroupImpl child, boolean broadcast) throws Exception
   {
      Node parentNode = utils.getGroupNode(session, parent);
      Node groupNode =
         parentNode.addNode(child.getGroupName(), JCROrganizationServiceImpl.JOS_HIERARCHY_GROUP_NODETYPE);

      String parentId = parent == null ? null : parent.getId();

      child.setParentId(parentId);
      child.setInternalId(groupNode.getUUID());

      if (broadcast)
      {
         preSave(child, true);
      }

      writeGroup(child, groupNode);
      session.save();

      putInCache(child);

      if (broadcast)
      {
         postSave(child, true);
      }
   }
1070012
@Override
    public void doSessionCreated(SessionManagementBean sessionBean) throws Exception {
        SessionMXBean sessionMxBean = managementServiceHandler.getSessionMXBean(sessionBean.getId());

        Map<String, String> userPrincipals = sessionBean.getUserPrincipalMap();

        // Send a notification for the principals for the logged-in session.
        if (userPrincipals != null) {
            Map<String, Map<String, String>> userData = new HashMap<>();
            userData.put(sessionMxBean.getObjectName().toString(), userPrincipals);

            Notification n2 = new Notification(SESSION_CREATED,
                    sessionMxBean,
                    managementServiceHandler.nextNotificationSequenceNumber(),
                    System.currentTimeMillis(),
                    "Session Credentials Registered");
            n2.setUserData(userData);
            sendNotification(n2);
        }
    }
649670
@SuppressWarnings("unchecked")
  private void commit() {
    long index = this.currentIndex;
    for (RaftSession session : sessions.getSessions(primitiveId)) {
      session.commit(index);
    }
  }
1141388
public void sessionIdle(IoSession session, IdleStatus status)
            throws Exception {
        SingleSessionIoHandler handler = (SingleSessionIoHandler) session
                .getAttribute(HANDLER);
        handler.sessionIdle(status);
    }
255750
public void addSession(String id, Session session) {
    synchronized (session) {
      SessionSchedulable schedulable =
        new SessionSchedulable(session, getType());
      idToSession.put(id, schedulable);
    }
  }
98673
public void onCandidate(long sessionId, int index, String id, String sdp) {
        send(new PeerConnectionActor.OnCandidate(sessionId, index, id, sdp));
    }
604900
long __getId() {
        Session session = this.wiki.__getHibernateSession();
        session.beginTransaction();
        session.lock(hibernateCategory, LockMode.NONE);
        long id = hibernateCategory.getId();
        session.getTransaction().commit();
        return id;
    }
360328
private void reassign() {
    if (member.type() == Member.Type.ACTIVE && !member.equals(context.getLeader())) {
      // Calculate this server's index within the collection of active members, excluding the leader.
      // This is done in a deterministic way by sorting the list of active members by ID.
      int index = 1;
      for (MemberState member : getActiveMemberStates((m1, m2) -> m1.getMember().id() - m2.getMember().id())) {
        if (!member.getMember().equals(context.getLeader())) {
          if (this.member.id() < member.getMember().id()) {
            index++;
          } else {
            break;
          }
        }
      }

      // Intersect the active members list with a sorted list of passive members to get assignments.
      List<MemberState> sortedPassiveMembers = getPassiveMemberStates((m1, m2) -> m1.getMember().id() - m2.getMember().id());
      assignedMembers = assignMembers(index, sortedPassiveMembers);
    } else {
      assignedMembers = new ArrayList<>(0);
    }
  }
1329328
public java.lang.String getLeaderMasterAddress() {
    java.lang.Object ref = leaderMasterAddress_;
    if (ref instanceof java.lang.String) {
      return (java.lang.String) ref;
    } else {
      com.google.protobuf.ByteString bs = 
          (com.google.protobuf.ByteString) ref;
      java.lang.String s = bs.toStringUtf8();
      if (bs.isValidUtf8()) {
        leaderMasterAddress_ = s;
      }
      return s;
    }
  }
59464
private void destroySessionByToken(String id) {
		System.out.println(new LogEntry("destroying session " + id));
		synchronized (sessions) {
			synchronized (sessionsMirror) {
				StandardSession session = (StandardSession) sessions.get(id);
				if (session != null) {
					session.onDestruction();
					sessions.remove(id);
					sessionsMirror.remove(id);
				}
			}
		}
	}
465611
public CompletableFuture<Void> anoint() {
    if (role.role() == RaftServer.Role.LEADER) {
      return CompletableFuture.completedFuture(null);
    }

    CompletableFuture<Void> future = new CompletableFuture<>();

    threadContext.execute(() -> {
      // Register a leader election listener to wait for the election of this node.
      Consumer<RaftMember> electionListener = new Consumer<RaftMember>() {
        @Override
        public void accept(RaftMember member) {
          if (member.memberId().equals(cluster.getMember().memberId())) {
            future.complete(null);
          } else {
            future.completeExceptionally(new RaftException.ProtocolException("Failed to transfer leadership"));
          }
          removeLeaderElectionListener(this);
        }
      };
      addLeaderElectionListener(electionListener);

      // If a leader already exists, request a leadership transfer from it. Otherwise, transition to the candidate
      // state and attempt to get elected.
      RaftMember member = getCluster().getMember();
      RaftMember leader = getLeader();
      if (leader != null) {
        protocol.transfer(leader.memberId(), TransferRequest.builder()
            .withMember(member.memberId())
            .build()).whenCompleteAsync((response, error) -> {
              if (error != null) {
                future.completeExceptionally(error);
              } else if (response.status() == RaftResponse.Status.ERROR) {
                future.completeExceptionally(response.error().createException());
              } else {
                transition(RaftServer.Role.CANDIDATE);
              }
            }, threadContext);
      } else {
        transition(RaftServer.Role.CANDIDATE);
      }
    });
    return future;
  }
166403
private Promise<PeerSession> pickCachedSession(int uid, final int keyGroupId) {
        return ManagedList.of(peerSessions.getValue(uid))
                .flatMap(PeerSessionsStorage.SESSIONS)
                .filter(PeerSession.BY_THEIR_GROUP(keyGroupId))
                .sorted(PeerSession.COMPARATOR)
                .firstPromise();
    }