@Override
    public void notifyOfHosts(
            int yourHostId,
            int[] hosts,
            SocketChannel[] sockets,
            SSLEngine[] sslEngines,
            InetSocketAddress listeningAddresses[],
            Map<Integer, JSONObject> jos) throws Exception {
        m_localHostId = yourHostId;
        long agreementHSId = getHSIdForLocalSite(AGREEMENT_SITE_ID);

        /*
         * Construct the set of agreement sites based on all the hosts that are connected
         */
        HashSet<Long> agreementSites = new HashSet<Long>();
        agreementSites.add(agreementHSId);

        m_network.start();//network must be running for register to work

        for (int ii = 0; ii < hosts.length; ii++) {
            networkLog.info(yourHostId + " notified of host " + hosts[ii]);
            agreementSites.add(CoreUtils.getHSIdFromHostAndSite(hosts[ii], AGREEMENT_SITE_ID));
            prepSocketChannel(sockets[ii]);
            ForeignHost fhost = null;
            try {
                fhost = new ForeignHost(this, hosts[ii], sockets[ii], m_config.deadHostTimeout,
                        listeningAddresses[ii], createPicoNetwork(sslEngines[ii], sockets[ii], false));
                putForeignHost(hosts[ii], fhost);
            } catch (java.io.IOException e) {
                org.voltdb.VoltDB.crashLocalVoltDB("Failed to instantiate foreign host", true, e);
            }
        }

        m_acceptor.accrue(jos);

        /*
         * Create the local agreement site. It knows that it is recovering because the number of
         * prexisting sites is > 0
         */
        SiteMailbox sm = new SiteMailbox(this, agreementHSId);
        createMailbox(agreementHSId, sm);
        m_agreementSite =
            new AgreementSite(
                    agreementHSId,
                    agreementSites,
                    yourHostId,
                    sm,
                    new InetSocketAddress(
                            m_config.zkInterface.split(":")[0],
                            Integer.parseInt(m_config.zkInterface.split(":")[1])),
                   m_config.backwardsTimeForgivenessWindow,
                   m_failedHostsCallback);

        /*
         * Now that the agreement site mailbox has been created it is safe
         * to enable read
         */
        for (ForeignHost fh : m_foreignHosts.values()) {
            fh.enableRead(VERBOTEN_THREADS);
        }
        m_agreementSite.start();

        /*
         * Do the usual thing of waiting for the agreement site
         * to join the cluster and creating the client
         */
        VERBOTEN_THREADS.addAll(m_network.getThreadIds());
        VERBOTEN_THREADS.addAll(m_agreementSite.getThreadIds());
        m_agreementSite.waitForRecovery();
        m_zk = org.voltcore.zk.ZKUtil.getClient(
                m_config.zkInterface, 60 * 1000, VERBOTEN_THREADS);
        if (m_zk == null) {
            throw new Exception("Timed out trying to connect local ZooKeeper instance");
        }

        /*
         * Publish the address of this node to ZK as seen by the leader
         * Also allows waitForGroupJoin to know the number of nodes in the cluster
         */
        HostInfo hostInfo;
        if (m_config.internalInterface.isEmpty()) {
            hostInfo = new HostInfo(new InetSocketAddress(m_joiner.m_reportedInternalInterface, m_config.internalPort).toString(),
                                    m_config.group, m_config.localSitesCount, m_config.recoveredPartitions);
        } else {
            hostInfo = new HostInfo(new InetSocketAddress(m_config.internalInterface, m_config.internalPort).toString(),
                                    m_config.group, m_config.localSitesCount, m_config.recoveredPartitions);
        }

        m_zk.create(CoreZK.hosts_host + getHostId(), hostInfo.toBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    }
@ Override public void notifyOfHosts ( int yourHostId , int [ ] hosts , SocketChannel [ ] sockets , SSLEngine [ ] sslEngines , InetSocketAddress listeningAddresses [ ] , Map < Integer , JSONObject > jos ) throws Exception { m_localHostId = yourHostId ; long agreementHSId = getHSIdForLocalSite ( AGREEMENT_SITE_ID ) ; HashSet < Long > agreementSites = new HashSet < Long > ( ) ; agreementSites . add ( agreementHSId ) ; m_network . start ( ) ; for ( int ii = 0 ; ii < hosts . length ; ii ++ ) { networkLog . info ( yourHostId + " notified of host " + hosts [ ii ] ) ; agreementSites . add ( CoreUtils . getHSIdFromHostAndSite ( hosts [ ii ] , AGREEMENT_SITE_ID ) ) ; prepSocketChannel ( sockets [ ii ] ) ; ForeignHost fhost = null ; try { fhost = new ForeignHost ( this , hosts [ ii ] , sockets [ ii ] , m_config . deadHostTimeout , listeningAddresses [ ii ] , createPicoNetwork ( sslEngines [ ii ] , sockets [ ii ] , false ) ) ; putForeignHost ( hosts [ ii ] , fhost ) ; } catch ( java . io . IOException e ) { org . voltdb . VoltDB . crashLocalVoltDB ( "Failed to instantiate foreign host" , true , e ) ; } } m_acceptor . accrue ( jos ) ; SiteMailbox sm = new SiteMailbox ( this , agreementHSId ) ; createMailbox ( agreementHSId , sm ) ; m_agreementSite = new AgreementSite ( agreementHSId , agreementSites , yourHostId , sm , new InetSocketAddress ( m_config . zkInterface . split ( ":" ) [ 0 ] , Integer . parseInt ( m_config . zkInterface . split ( ":" ) [ 1 ] ) ) , m_config . backwardsTimeForgivenessWindow , m_failedHostsCallback ) ; for ( ForeignHost fh : m_foreignHosts . values ( ) ) { fh . enableRead ( VERBOTEN_THREADS ) ; } m_agreementSite . start ( ) ; VERBOTEN_THREADS . addAll ( m_network . getThreadIds ( ) ) ; VERBOTEN_THREADS . addAll ( m_agreementSite . getThreadIds ( ) ) ; m_agreementSite . waitForRecovery ( ) ; m_zk = org . voltcore . zk . ZKUtil . getClient ( m_config . zkInterface , 60 * 1000 , VERBOTEN_THREADS ) ; if ( m_zk == null ) { throw new Exception ( "Timed out trying to connect local ZooKeeper instance" ) ; } HostInfo hostInfo ; if ( m_config . internalInterface . isEmpty ( ) ) { hostInfo = new HostInfo ( new InetSocketAddress ( m_joiner . m_reportedInternalInterface , m_config . internalPort ) . toString ( ) , m_config . group , m_config . localSitesCount , m_config . recoveredPartitions ) ; } else { hostInfo = new HostInfo ( new InetSocketAddress ( m_config . internalInterface , m_config . internalPort ) . toString ( ) , m_config . group , m_config . localSitesCount , m_config . recoveredPartitions ) ; } m_zk . create ( CoreZK . hosts_host + getHostId ( ) , hostInfo . toBytes ( ) , Ids . OPEN_ACL_UNSAFE , CreateMode . EPHEMERAL ) ; }
/*
SJ invokes this method after a node finishes connecting to the entire cluster.
This method constructs all the hosts and puts them in the map