@ Override @ Synchronized public void takeLeadership ( CuratorFramework client ) throws Exception { log . info ( "Obtained leadership to monitor the Host to Segment Container Mapping" ) ; //Attempt a rebalance whenever leadership is obtained to ensure no host events are missed. hostsChange . release ( ) ; //Start cluster monitor. pravegaServiceCluster = new ClusterZKImpl ( client , ClusterType . HOST ) ; //Add listener to track host changes on the monitored pravega cluster. pravegaServiceCluster . addListener ( ( type , host ) -> { switch ( type ) { case HOST_ADDED : case HOST_REMOVED : //We don't keep track of the hosts and we always query for the entire set from the cluster //when changes occur. This is to avoid any inconsistencies if we miss any notifications. log . info ( "Received event: {} for host: {}. Wake up leader for rebalancing" , type , host ) ; hostsChange . release ( ) ; break ; case ERROR : //This event should be due to ZK connection errors and would have been received by the monitor too, //hence not handling it explicitly here. log . info ( "Received error event when monitoring the pravega host cluster, ignoring..." ) ; break ; } } ) ; //Keep looping here as long as possible to stay as the leader and exclusively monitor the pravega cluster. while ( true ) { try { if ( suspended . get ( ) ) { log . info ( "Monitor is suspended, waiting for notification to resume" ) ; suspendMonitor . acquire ( ) ; log . info ( "Resuming monitor" ) ; } hostsChange . acquire ( ) ; log . info ( "Received rebalance event" ) ; // Wait here until rebalance can be performed. waitForRebalance ( ) ; // Clear all events that has been received until this point since this will be included in the current // rebalance operation. hostsChange . drainPermits ( ) ; triggerRebalance ( ) ; } catch ( InterruptedException e ) { log . warn ( "Leadership interrupted, releasing monitor thread" ) ; //Stop watching the pravega cluster. pravegaServiceCluster . close ( ) ; throw e ; } catch ( Exception e ) { //We will not release leadership if in suspended mode. if ( ! suspended . get ( ) ) { log . warn ( "Failed to perform rebalancing, relinquishing leadership" ) ; //Stop watching the pravega cluster. pravegaServiceCluster . close ( ) ; throw e ; } } } }
This function is called when the current instance is made the leader . The leadership is relinquished when this function exits .