void waitNewLeaderAckFromQuorum ( ) throws TimeoutException , InterruptedException , IOException { LOG . debug ( "Waiting for synchronization to followers complete." ) ; int completeCount = 0 ; Zxid lastZxid = persistence . getLatestZxid ( ) ; while ( completeCount < this . quorumMap . size ( ) ) { // Here we should use sync_timeout. MessageTuple tuple = filter . getExpectedMessage ( MessageType . ACK , null , getSyncTimeoutMs ( ) ) ; ZabMessage . Ack ack = tuple . getMessage ( ) . getAck ( ) ; String source = tuple . getServerId ( ) ; Zxid zxid = MessageBuilder . fromProtoZxid ( ack . getZxid ( ) ) ; if ( ! this . quorumMap . containsKey ( source ) ) { LOG . warn ( "Quorum set doesn't contain {}, a bug?" , source ) ; continue ; } if ( zxid . compareTo ( lastZxid ) != 0 ) { LOG . error ( "The follower {} is not correctly synchronized." , source ) ; throw new RuntimeException ( "The synchronized follower's last zxid" + "doesn't match last zxid of current leader." ) ; } PeerHandler ph = this . quorumMap . get ( source ) ; ph . setLastAckedZxid ( zxid ) ; completeCount ++ ; } }
Waits for synchronization to followers complete .