protected boolean notifyOnKill(Set<Long> hsIds, Map<Long, Long> decision) {

        SiteFailureMessage.Builder sfmb = SiteFailureMessage.
                builder()
                .decisions(decision.keySet())
                .failures(decision.keySet());

        Set<Long> dests = Sets.filter(m_seeker.getSurvivors(), not(equalTo(m_hsId)));
        if (dests.isEmpty()) return true;

        sfmb.survivors(Sets.difference(m_seeker.getSurvivors(), decision.keySet()));
        sfmb.safeTxnIds(getSafeTxnIdsForSites(hsIds));

        SiteFailureMessage sfm = sfmb.build();
        m_mailbox.send(Longs.toArray(dests), sfm);

        m_recoveryLog.info("Agreement, Sending ["
                + CoreUtils.hsIdCollectionToString(dests) + "]  " + sfm);

        // Check to see we've made the same decision before, if so, it's likely
        // that we've entered a loop, exit here.
        if (m_localHistoricDecisions.size() >= 100) {
            // Too many decisions have been made without converging
            RateLimitedLogger.tryLogForMessage(System.currentTimeMillis(),
                                               10, TimeUnit.SECONDS,
                                               m_recoveryLog,
                                               Level.WARN,
                                               "Agreement, %d local decisions have been made without converging",
                                               m_localHistoricDecisions.size());
        }
        for (SiteFailureMessage lhd : m_localHistoricDecisions) {
            if (lhd.m_survivors.equals(sfm.m_survivors)) {
                m_recoveryLog.info("Agreement, detected decision loop. Exiting");
                return true;
            }
        }
        m_localHistoricDecisions.add(sfm);

        // Wait for all survivors in the local decision to send their decisions over.
        // If one of the host's decision conflicts with ours, remove that host's link
        // and repeat the decision process.
        final Set<Long> expectedSurvivors = Sets.filter(sfm.m_survivors, not(equalTo(m_hsId)));
        m_recoveryLog.info("Agreement, Waiting for agreement on decision from survivors " +
                           CoreUtils.hsIdCollectionToString(expectedSurvivors));

        final Iterator<SiteFailureMessage> iter = m_decidedSurvivors.values().iterator();
        while (iter.hasNext()) {
            final SiteFailureMessage remoteDecision = iter.next();
            if (expectedSurvivors.contains(remoteDecision.m_sourceHSId)) {
                if (remoteDecision.m_decision.contains(m_hsId)) {
                    iter.remove();
                    m_recoveryLog.info("Agreement, Received inconsistent decision from " +
                                       CoreUtils.hsIdToString(remoteDecision.m_sourceHSId) + ", " + remoteDecision);
                    final FaultMessage localFault = new FaultMessage(m_hsId, remoteDecision.m_sourceHSId);
                    localFault.m_sourceHSId = m_hsId;
                    m_mailbox.deliverFront(localFault);
                    return false;
                }
            }
        }

        long start = System.currentTimeMillis();
        boolean allDecisionsMatch = true;
        do {
            final VoltMessage msg = m_mailbox.recvBlocking(receiveSubjects, 5);
            if (msg == null) {
                // Send a heartbeat to keep the dead host timeout active.
                m_meshAide.sendHeartbeats(m_seeker.getSurvivors());
                final long duration = System.currentTimeMillis() - start;
                if (duration > 20000) {
                    m_recoveryLog.error("Agreement, Still waiting for decisions from " +
                                        CoreUtils.hsIdCollectionToString(Sets.difference(expectedSurvivors, m_decidedSurvivors.keySet())) +
                                        " after " + TimeUnit.MILLISECONDS.toSeconds(duration) + " seconds");
                    start = System.currentTimeMillis();
                }
                continue;
            }

            if (m_hsId != msg.m_sourceHSId && !expectedSurvivors.contains(msg.m_sourceHSId)) {
                // Ignore messages from failed sites
                continue;
            }

            if (msg.getSubject() == Subject.SITE_FAILURE_UPDATE.getId()) {
                final SiteFailureMessage fm = (SiteFailureMessage) msg;
                if (!fm.m_decision.isEmpty()) {
                    if (expectedSurvivors.contains(fm.m_sourceHSId)) {
                        if (fm.m_decision.contains(m_hsId)) {
                            m_decidedSurvivors.remove(fm.m_sourceHSId);
                            // The remote host has decided that we are gone, remove the remote host
                            final FaultMessage localFault = new FaultMessage(m_hsId, fm.m_sourceHSId);
                            localFault.m_sourceHSId = m_hsId;
                            m_mailbox.deliverFront(localFault);
                            return false;
                        } else {
                            m_decidedSurvivors.put(fm.m_sourceHSId, fm);
                        }
                    }
                } else {
                    m_mailbox.deliverFront(fm);
                    return false;
                }
            } else if (msg.getSubject() == Subject.FAILURE.getId()) {
                final FaultMessage fm = (FaultMessage) msg;
                if (!fm.decided) {
                    // In case of concurrent fault, handle it
                    m_mailbox.deliverFront(msg);
                    return false;
                }

                /* So this is a final update message.
                 * For every final update message this node expects to receive, two messages will be
                 * actually send to ZookeeperServer thread. One is a fault message which is used to
                 * trigger a new round of message broadcasting, another is the original final update
                 * message which is used by mesh arbiter to reach the agreement.
                 */

                // Send the message to ZookeeperServer thread only if the final update message can give
                // us new information (new alive/dead host)
                if (!m_seeker.alreadyKnow(fm) && mayIgnore(hsIds, fm) == Discard.DoNot) {
                    m_mailbox.deliverFront(msg);
                    return false;
                }

            }

            for (SiteFailureMessage remoteDecision : m_decidedSurvivors.values()) {
                if (!sfm.m_survivors.equals(remoteDecision.m_survivors)) {
                    allDecisionsMatch = false;
                }
            }
        } while (!m_decidedSurvivors.keySet().containsAll(expectedSurvivors) && allDecisionsMatch);

        return true;
    }
protected boolean notifyOnKill ( Set < Long > hsIds , Map < Long , Long > decision ) { SiteFailureMessage . Builder sfmb = SiteFailureMessage . builder ( ) . decisions ( decision . keySet ( ) ) . failures ( decision . keySet ( ) ) ; Set < Long > dests = Sets . filter ( m_seeker . getSurvivors ( ) , not ( equalTo ( m_hsId ) ) ) ; if ( dests . isEmpty ( ) ) return true ; sfmb . survivors ( Sets . difference ( m_seeker . getSurvivors ( ) , decision . keySet ( ) ) ) ; sfmb . safeTxnIds ( getSafeTxnIdsForSites ( hsIds ) ) ; SiteFailureMessage sfm = sfmb . build ( ) ; m_mailbox . send ( Longs . toArray ( dests ) , sfm ) ; m_recoveryLog . info ( "Agreement, Sending [" + CoreUtils . hsIdCollectionToString ( dests ) + "]  " + sfm ) ; if ( m_localHistoricDecisions . size ( ) >= 100 ) { RateLimitedLogger . tryLogForMessage ( System . currentTimeMillis ( ) , 10 , TimeUnit . SECONDS , m_recoveryLog , Level . WARN , "Agreement, %d local decisions have been made without converging" , m_localHistoricDecisions . size ( ) ) ; } for ( SiteFailureMessage lhd : m_localHistoricDecisions ) { if ( lhd . m_survivors . equals ( sfm . m_survivors ) ) { m_recoveryLog . info ( "Agreement, detected decision loop. Exiting" ) ; return true ; } } m_localHistoricDecisions . add ( sfm ) ; final Set < Long > expectedSurvivors = Sets . filter ( sfm . m_survivors , not ( equalTo ( m_hsId ) ) ) ; m_recoveryLog . info ( "Agreement, Waiting for agreement on decision from survivors " + CoreUtils . hsIdCollectionToString ( expectedSurvivors ) ) ; final Iterator < SiteFailureMessage > iter = m_decidedSurvivors . values ( ) . iterator ( ) ; while ( iter . hasNext ( ) ) { final SiteFailureMessage remoteDecision = iter . next ( ) ; if ( expectedSurvivors . contains ( remoteDecision . m_sourceHSId ) ) { if ( remoteDecision . m_decision . contains ( m_hsId ) ) { iter . remove ( ) ; m_recoveryLog . info ( "Agreement, Received inconsistent decision from " + CoreUtils . hsIdToString ( remoteDecision . m_sourceHSId ) + ", " + remoteDecision ) ; final FaultMessage localFault = new FaultMessage ( m_hsId , remoteDecision . m_sourceHSId ) ; localFault . m_sourceHSId = m_hsId ; m_mailbox . deliverFront ( localFault ) ; return false ; } } } long start = System . currentTimeMillis ( ) ; boolean allDecisionsMatch = true ; do { final VoltMessage msg = m_mailbox . recvBlocking ( receiveSubjects , 5 ) ; if ( msg == null ) { m_meshAide . sendHeartbeats ( m_seeker . getSurvivors ( ) ) ; final long duration = System . currentTimeMillis ( ) - start ; if ( duration > 20000 ) { m_recoveryLog . error ( "Agreement, Still waiting for decisions from " + CoreUtils . hsIdCollectionToString ( Sets . difference ( expectedSurvivors , m_decidedSurvivors . keySet ( ) ) ) + " after " + TimeUnit . MILLISECONDS . toSeconds ( duration ) + " seconds" ) ; start = System . currentTimeMillis ( ) ; } continue ; } if ( m_hsId != msg . m_sourceHSId && ! expectedSurvivors . contains ( msg . m_sourceHSId ) ) { continue ; } if ( msg . getSubject ( ) == Subject . SITE_FAILURE_UPDATE . getId ( ) ) { final SiteFailureMessage fm = ( SiteFailureMessage ) msg ; if ( ! fm . m_decision . isEmpty ( ) ) { if ( expectedSurvivors . contains ( fm . m_sourceHSId ) ) { if ( fm . m_decision . contains ( m_hsId ) ) { m_decidedSurvivors . remove ( fm . m_sourceHSId ) ; final FaultMessage localFault = new FaultMessage ( m_hsId , fm . m_sourceHSId ) ; localFault . m_sourceHSId = m_hsId ; m_mailbox . deliverFront ( localFault ) ; return false ; } else { m_decidedSurvivors . put ( fm . m_sourceHSId , fm ) ; } } } else { m_mailbox . deliverFront ( fm ) ; return false ; } } else if ( msg . getSubject ( ) == Subject . FAILURE . getId ( ) ) { final FaultMessage fm = ( FaultMessage ) msg ; if ( ! fm . decided ) { m_mailbox . deliverFront ( msg ) ; return false ; } if ( ! m_seeker . alreadyKnow ( fm ) && mayIgnore ( hsIds , fm ) == Discard . DoNot ) { m_mailbox . deliverFront ( msg ) ; return false ; } } for ( SiteFailureMessage remoteDecision : m_decidedSurvivors . values ( ) ) { if ( ! sfm . m_survivors . equals ( remoteDecision . m_survivors ) ) { allDecisionsMatch = false ; } } } while ( ! m_decidedSurvivors . keySet ( ) . containsAll ( expectedSurvivors ) && allDecisionsMatch ) ; return true ; }
Notify all survivors when you are closing links to nodes
@param decision map where the keys contain the kill sites
and its values are their last known safe transaction ids
@return true if successfully confirmed that all survivors
agree on the decision, false otherwise.