private void initiateSPIMigrationIfRequested ( Iv2InitiateTaskMessage msg ) { if ( ! "@MigratePartitionLeader" . equals ( msg . getStoredProcedureName ( ) ) ) { return ; } final Object [ ] params = msg . getParameters ( ) ; int pid = Integer . parseInt ( params [ 1 ] . toString ( ) ) ; if ( pid != m_partitionId ) { tmLog . warn ( String . format ( "@MigratePartitionLeader executed at a wrong partition %d for partition %d." , m_partitionId , pid ) ) ; return ; } RealVoltDB db = ( RealVoltDB ) VoltDB . instance ( ) ; int hostId = Integer . parseInt ( params [ 2 ] . toString ( ) ) ; Long newLeaderHSId = db . getCartographer ( ) . getHSIDForPartitionHost ( hostId , pid ) ; if ( newLeaderHSId == null || newLeaderHSId == m_hsId ) { tmLog . warn ( String . format ( "@MigratePartitionLeader the partition leader is already on the host %d or the host id is invalid." , hostId ) ) ; return ; } SpScheduler scheduler = ( SpScheduler ) m_scheduler ; scheduler . checkPointMigratePartitionLeader ( ) ; scheduler . m_isLeader = false ; m_newLeaderHSID = newLeaderHSId ; m_migratePartitionLeaderStatus = MigratePartitionLeaderStatus . STARTED ; LeaderCache leaderAppointee = new LeaderCache ( m_messenger . getZK ( ) , "initiateSPIMigrationIfRequested-" + m_partitionId , VoltZK . iv2appointees ) ; try { leaderAppointee . start ( true ) ; leaderAppointee . put ( pid , LeaderCache . suffixHSIdsWithMigratePartitionLeaderRequest ( newLeaderHSId ) ) ; } catch ( InterruptedException | ExecutionException | KeeperException e ) { VoltDB . crashLocalVoltDB ( "fail to start MigratePartitionLeader" , true , e ) ; } finally { try { leaderAppointee . shutdown ( ) ; } catch ( InterruptedException e ) { } } tmLog . info ( "MigratePartitionLeader for partition " + pid + " to " + CoreUtils . hsIdToString ( newLeaderHSId ) ) ; //notify the new leader right away if the current leader has drained all transactions. notifyNewLeaderOfTxnDoneIfNeeded ( ) ; }
rerouted from this moment on until the transactions are correctly routed to new leader .