+ private void initializeDownPeerMemberNamesFromClusterState() {
+ java.util.Optional<Cluster> cluster = getRaftActorContext().getCluster();
+ if (!cluster.isPresent()) {
+ return;
+ }
+
+ CurrentClusterState state = cluster.get().state();
+ Set<Member> unreachable = state.getUnreachable();
+
+ LOG.debug(
+ "{}: initializeDownPeerMemberNamesFromClusterState - current downPeerMemberNames: {}, unreachable: {}",
+ persistenceId(), downPeerMemberNames, unreachable);
+
+ downPeerMemberNames.clear();
+ for (Member m: unreachable) {
+ downPeerMemberNames.add(MemberName.forName(m.getRoles().iterator().next()));
+ }
+
+ for (Member m: state.getMembers()) {
+ if (m.status() != MemberStatus.up() && m.status() != MemberStatus.weaklyUp()) {
+ LOG.debug("{}: Adding down member with status {}", persistenceId(), m.status());
+ downPeerMemberNames.add(MemberName.forName(m.getRoles().iterator().next()));
+ }
+ }
+
+ LOG.debug("{}: new downPeerMemberNames: {}", persistenceId(), downPeerMemberNames);
+ }
+
+ private void onCandidateRemoved(final CandidateRemoved message) {
+ LOG.debug("{}: onCandidateRemoved: {}", persistenceId(), message);
+
+ if (isLeader()) {
+ String currentOwner = getCurrentOwner(message.getEntityPath());
+ writeNewOwner(message.getEntityPath(),
+ newOwner(currentOwner, message.getRemainingCandidates(),
+ getEntityOwnerElectionStrategy(message.getEntityPath())));
+ }
+ }
+
+ private EntityOwnerSelectionStrategy getEntityOwnerElectionStrategy(final YangInstanceIdentifier entityPath) {
+ final String entityType = EntityOwnersModel.entityTypeFromEntityPath(entityPath);
+ return strategyConfig.createStrategy(entityType, entityOwnershipStatistics.byEntityType(entityType));
+ }
+
+ private void onCandidateAdded(final CandidateAdded message) {
+ if (!isLeader()) {
+ return;
+ }
+
+ LOG.debug("{}: onCandidateAdded: {}", persistenceId(), message);
+
+ // Since a node's candidate member is only added by the node itself, we can assume the node is up so
+ // remove it from the downPeerMemberNames.
+ downPeerMemberNames.remove(MemberName.forName(message.getNewCandidate()));
+
+ final String currentOwner = getCurrentOwner(message.getEntityPath());
+ final EntityOwnerSelectionStrategy strategy = getEntityOwnerElectionStrategy(message.getEntityPath());
+
+ // Available members is all the known peers - the number of peers that are down + self
+ // So if there are 2 peers and 1 is down then availableMembers will be 2
+ final int availableMembers = getRaftActorContext().getPeerIds().size() - downPeerMemberNames.size() + 1;
+
+ LOG.debug("{}: Using strategy {} to select owner, currentOwner = {}", persistenceId(), strategy, currentOwner);
+
+ if (strategy.getSelectionDelayInMillis() == 0L) {
+ writeNewOwner(message.getEntityPath(), newOwner(currentOwner, message.getAllCandidates(),
+ strategy));
+ } else if (message.getAllCandidates().size() == availableMembers) {
+ LOG.debug("{}: Received the maximum candidates requests : {} writing new owner",
+ persistenceId(), availableMembers);
+ cancelOwnerSelectionTask(message.getEntityPath());
+ writeNewOwner(message.getEntityPath(), newOwner(currentOwner, message.getAllCandidates(),
+ strategy));
+ } else {
+ scheduleOwnerSelection(message.getEntityPath(), message.getAllCandidates(), strategy);
+ }
+ }
+
+ private void onPeerDown(final PeerDown peerDown) {
+ LOG.info("{}: onPeerDown: {}", persistenceId(), peerDown);
+
+ MemberName downMemberName = peerDown.getMemberName();
+ if (downPeerMemberNames.add(downMemberName) && isLeader()) {
+ // Select new owners for entities owned by the down peer and which have other candidates. For an entity for
+ // which the down peer is the only candidate, we leave it as the owner and don't clear it. This is done to
+ // handle the case where the peer member process is actually still running but the node is partitioned.
+ // When the partition is healed, the peer just remains as the owner. If the peer process actually restarted,
+ // it will first remove all its candidates on startup. If another candidate is registered during the time
+ // the peer is down, the new candidate will be selected as the new owner.
+
+ selectNewOwnerForEntitiesOwnedBy(ImmutableSet.of(downMemberName.getName()));
+ }
+ }
+
+ private void selectNewOwnerForEntitiesOwnedBy(final Set<String> ownedBy) {
+ final List<Modification> modifications = new ArrayList<>();
+ searchForEntitiesOwnedBy(ownedBy, (entityTypeNode, entityNode) -> {
+ YangInstanceIdentifier entityPath = YangInstanceIdentifier.builder(ENTITY_TYPES_PATH)
+ .node(entityTypeNode.getIdentifier()).node(ENTITY_NODE_ID).node(entityNode.getIdentifier())
+ .node(ENTITY_OWNER_NODE_ID).build();
+ String newOwner = newOwner(getCurrentOwner(entityPath), getCandidateNames(entityNode),
+ getEntityOwnerElectionStrategy(entityPath));
+
+ if (!newOwner.isEmpty()) {
+ LOG.debug("{}: Found entity {}, writing new owner {}", persistenceId(), entityPath, newOwner);
+
+ modifications.add(new WriteModification(entityPath,
+ ImmutableNodes.leafNode(ENTITY_OWNER_NODE_ID, newOwner)));
+
+ } else {
+ LOG.debug("{}: Found entity {} but no other candidates - not clearing owner", persistenceId(),
+ entityPath, newOwner);
+ }
+ });
+
+ commitCoordinator.commitModifications(modifications, this);
+ }
+
+ private void onPeerUp(final PeerUp peerUp) {
+ LOG.debug("{}: onPeerUp: {}", persistenceId(), peerUp);
+
+ downPeerMemberNames.remove(peerUp.getMemberName());
+
+ // Notify the coordinator to check if pending modifications need to be sent. We do this here
+ // to handle the case where the leader's peer address isn't known yet when a prior state or
+ // leader change occurred.
+ commitCoordinator.onStateChanged(this, isLeader());
+
+ if (isLeader()) {
+ // Try to assign owners for entities that have no current owner. It's possible the peer that is now up
+ // had previously registered as a candidate and was the only candidate but the owner write tx couldn't be
+ // committed due to a leader change. Eg, the leader is able to successfully commit the candidate add tx but
+ // becomes isolated before it can commit the owner change and switches to follower. The majority partition
+ // with a new leader has the candidate but the entity has no owner. When the partition is healed and the
+ // previously isolated leader reconnects, we'll receive onPeerUp and, if there's still no owner, the
+ // previous leader will gain ownership.
+ selectNewOwnerForEntitiesOwnedBy(ImmutableSet.of(""));
+ }
+ }
+
+ private static Collection<String> getCandidateNames(final MapEntryNode entity) {
+ Collection<MapEntryNode> candidates = ((MapNode)entity.getChild(CANDIDATE_NODE_ID).get()).getValue();
+ Collection<String> candidateNames = new ArrayList<>(candidates.size());
+ for (MapEntryNode candidate: candidates) {
+ candidateNames.add(candidate.getChild(CANDIDATE_NAME_NODE_ID).get().getValue().toString());
+ }
+
+ return candidateNames;
+ }
+
+ private void searchForEntitiesOwnedBy(final Set<String> ownedBy, final EntityWalker walker) {
+ LOG.debug("{}: Searching for entities owned by {}", persistenceId(), ownedBy);
+
+ searchForEntities((entityTypeNode, entityNode) -> {
+ Optional<DataContainerChild<? extends PathArgument, ?>> possibleOwner =
+ entityNode.getChild(ENTITY_OWNER_NODE_ID);
+ String currentOwner = possibleOwner.isPresent() ? possibleOwner.get().getValue().toString() : "";
+ if (ownedBy.contains(currentOwner)) {
+ walker.onEntity(entityTypeNode, entityNode);
+ }
+ });
+ }
+
+ private void removeCandidateFromEntities(final MemberName member) {
+ final List<Modification> modifications = new ArrayList<>();
+ searchForEntities((entityTypeNode, entityNode) -> {
+ if (hasCandidate(entityNode, member)) {
+ YangInstanceIdentifier entityId =
+ (YangInstanceIdentifier) entityNode.getIdentifier().getKeyValues().get(ENTITY_ID_QNAME);
+ YangInstanceIdentifier candidatePath = candidatePath(
+ entityTypeNode.getIdentifier().getKeyValues().get(ENTITY_TYPE_QNAME).toString(),
+ entityId, member.getName());
+
+ LOG.info("{}: Found entity {}, removing candidate {}, path {}", persistenceId(), entityId,
+ member, candidatePath);
+
+ modifications.add(new DeleteModification(candidatePath));
+ }
+ });
+
+ commitCoordinator.commitModifications(modifications, this);
+ }
+
+ private static boolean hasCandidate(final MapEntryNode entity, final MemberName candidateName) {
+ return ((MapNode)entity.getChild(CANDIDATE_NODE_ID).get()).getChild(candidateNodeKey(candidateName.getName()))
+ .isPresent();
+ }
+
+ private void searchForEntities(final EntityWalker walker) {
+ Optional<NormalizedNode<?, ?>> possibleEntityTypes = getDataStore().readNode(ENTITY_TYPES_PATH);
+ if (!possibleEntityTypes.isPresent()) {
+ return;
+ }
+
+ for (MapEntryNode entityType: ((MapNode) possibleEntityTypes.get()).getValue()) {
+ Optional<DataContainerChild<?, ?>> possibleEntities = entityType.getChild(ENTITY_NODE_ID);
+ if (!possibleEntities.isPresent()) {
+ // shouldn't happen but handle anyway
+ continue;
+ }
+
+ for (MapEntryNode entity: ((MapNode) possibleEntities.get()).getValue()) {
+ walker.onEntity(entityType, entity);
+ }
+ }
+ }
+
+ private void writeNewOwner(final YangInstanceIdentifier entityPath, final String newOwner) {
+ LOG.debug("{}: Writing new owner {} for entity {}", persistenceId(), newOwner, entityPath);
+
+ commitCoordinator.commitModification(new WriteModification(entityPath.node(ENTITY_OWNER_QNAME),
+ ImmutableNodes.leafNode(ENTITY_OWNER_NODE_ID, newOwner)), this);
+ }
+
+ /**
+ * Schedule a new owner selection job. Cancelling any outstanding job if it has not been cancelled.
+ */
+ private void scheduleOwnerSelection(final YangInstanceIdentifier entityPath, final Collection<String> allCandidates,
+ final EntityOwnerSelectionStrategy strategy) {
+ cancelOwnerSelectionTask(entityPath);
+
+ LOG.debug("{}: Scheduling owner selection after {} ms", persistenceId(), strategy.getSelectionDelayInMillis());
+
+ final Cancellable lastScheduledTask = context().system().scheduler().scheduleOnce(
+ FiniteDuration.apply(strategy.getSelectionDelayInMillis(), TimeUnit.MILLISECONDS), self(),
+ new SelectOwner(entityPath, allCandidates, strategy), context().system().dispatcher(), self());
+
+ entityToScheduledOwnershipTask.put(entityPath, lastScheduledTask);
+ }
+
+ private void cancelOwnerSelectionTask(final YangInstanceIdentifier entityPath) {
+ final Cancellable lastScheduledTask = entityToScheduledOwnershipTask.get(entityPath);
+ if (lastScheduledTask != null && !lastScheduledTask.isCancelled()) {
+ lastScheduledTask.cancel();
+ }
+ }
+
+ private String newOwner(final String currentOwner, final Collection<String> candidates,
+ final EntityOwnerSelectionStrategy ownerSelectionStrategy) {
+ Collection<String> viableCandidates = getViableCandidates(candidates);
+ if (viableCandidates.isEmpty()) {
+ return "";
+ }
+ return ownerSelectionStrategy.newOwner(currentOwner, viableCandidates);
+ }
+
+ private Collection<String> getViableCandidates(final Collection<String> candidates) {
+ Collection<String> viableCandidates = new ArrayList<>();
+
+ for (String candidate : candidates) {
+ if (!downPeerMemberNames.contains(MemberName.forName(candidate))) {
+ viableCandidates.add(candidate);
+ }
+ }
+ return viableCandidates;
+ }
+
+ private String getCurrentOwner(final YangInstanceIdentifier entityId) {
+ Optional<NormalizedNode<?, ?>> optionalEntityOwner = getDataStore().readNode(entityId.node(ENTITY_OWNER_QNAME));
+ if (optionalEntityOwner.isPresent()) {
+ return optionalEntityOwner.get().getValue().toString();
+ }
+ return null;