@ Override public void join ( String peer ) throws Exception { LOG . debug ( "Follower joins in." ) ; try { LOG . debug ( "Query leader from {}" , peer ) ; Message query = MessageBuilder . buildQueryLeader ( ) ; MessageTuple tuple = null ; while ( true ) { // The joiner might gets started before the cluster gets started. // Instead of reporting failure immediately, the follower will retry // joining the cluster until joining the cluster successfully. try { sendMessage ( peer , query ) ; tuple = filter . getExpectedMessage ( MessageType . QUERY_LEADER_REPLY , peer , config . getTimeoutMs ( ) ) ; break ; } catch ( TimeoutException ex ) { long retryInterval = 1 ; LOG . warn ( "Timeout while contacting leader, going to retry after {} " + "second." , retryInterval ) ; // Waits for 1 second. Thread . sleep ( retryInterval * 1000 ) ; } } this . electedLeader = tuple . getMessage ( ) . getReply ( ) . getLeader ( ) ; LOG . debug ( "Got current leader {}" , this . electedLeader ) ; /* -- Synchronizing phase -- */ while ( true ) { try { joinSynchronization ( ) ; break ; } catch ( TimeoutException | BackToElectionException ex ) { LOG . debug ( "Timeout({} ms) in synchronizing, retrying. Last zxid : {}" , getSyncTimeoutMs ( ) , persistence . getLatestZxid ( ) ) ; transport . clear ( electedLeader ) ; clearMessageQueue ( ) ; } } // See if it can be restored from the snapshot file. restoreFromSnapshot ( ) ; // Delivers all transactions in log before entering broadcasting phase. deliverUndeliveredTxns ( ) ; /* -- Broadcasting phase -- */ // Initialize the vote for leader election. this . election . specifyLeader ( this . electedLeader ) ; changePhase ( Phase . BROADCASTING ) ; accepting ( ) ; } catch ( InterruptedException e ) { LOG . debug ( "Participant is canceled by user." ) ; throw e ; } catch ( TimeoutException e ) { LOG . debug ( "Didn't hear message from {} for {} milliseconds. Going" + " back to leader election." , this . electedLeader , this . config . getTimeoutMs ( ) ) ; if ( persistence . getLastSeenConfig ( ) == null ) { throw new JoinFailure ( "Fails to join cluster." ) ; } } catch ( BackToElectionException e ) { LOG . debug ( "Got GO_BACK message from queue, going back to electing." ) ; if ( persistence . getLastSeenConfig ( ) == null ) { throw new JoinFailure ( "Fails to join cluster." ) ; } } catch ( LeftCluster e ) { LOG . debug ( "Exit running : {}" , e . getMessage ( ) ) ; throw e ; } catch ( Exception e ) { LOG . error ( "Caught exception" , e ) ; throw e ; } finally { changePhase ( Phase . FINALIZING ) ; } }
Starts from joining some one who is in cluster ..