void broadcasting ( ) throws TimeoutException , InterruptedException , IOException , ExecutionException { // Initialization. broadcastingInit ( ) ; try { while ( this . quorumMap . size ( ) >= clusterConfig . getQuorumSize ( ) ) { MessageTuple tuple = filter . getMessage ( config . getTimeoutMs ( ) ) ; Message msg = tuple . getMessage ( ) ; String source = tuple . getServerId ( ) ; // Checks if it's DISCONNECTED message. if ( msg . getType ( ) == MessageType . DISCONNECTED ) { String peerId = msg . getDisconnected ( ) . getServerId ( ) ; if ( quorumMap . containsKey ( peerId ) ) { onDisconnected ( tuple ) ; } else { this . transport . clear ( peerId ) ; } continue ; } if ( ! quorumMap . containsKey ( source ) ) { // Received a message sent from a peer who is outside the quorum. handleMessageOutsideQuorum ( tuple ) ; } else { // Received a message sent from the peer who is in quorum. handleMessageFromQuorum ( tuple ) ; PeerHandler ph = quorumMap . get ( source ) ; if ( ph != null ) { ph . updateHeartbeatTime ( ) ; } checkFollowerLiveness ( ) ; } } LOG . debug ( "Detects the size of the ensemble is less than the" + "quorum size {}, goes back to electing phase." , getQuorumSize ( ) ) ; } finally { ackProcessor . shutdown ( ) ; preProcessor . shutdown ( ) ; commitProcessor . shutdown ( ) ; syncProcessor . shutdown ( ) ; snapProcessor . shutdown ( ) ; this . lastDeliveredZxid = commitProcessor . getLastDeliveredZxid ( ) ; this . participantState . updateLastDeliveredZxid ( this . lastDeliveredZxid ) ; } }
Entering broadcasting phase leader broadcasts proposal to followers .