@ Override public ManagerStatusResponse handleStatusRequest ( int viewId ) { try { // As long as we have an older view, we can still process topologies from the old coordinator waitForView ( viewId , getGlobalTimeout ( ) , TimeUnit . MILLISECONDS ) ; } catch ( InterruptedException e ) { // Shutting down, send back an empty status Thread . currentThread ( ) . interrupt ( ) ; return new ManagerStatusResponse ( Collections . emptyMap ( ) , true ) ; } Map < String , CacheStatusResponse > caches = new HashMap <> ( ) ; synchronized ( runningCaches ) { latestStatusResponseViewId = viewId ; for ( Map . Entry < String , LocalCacheStatus > e : runningCaches . entrySet ( ) ) { String cacheName = e . getKey ( ) ; LocalCacheStatus cacheStatus = runningCaches . get ( cacheName ) ; caches . put ( e . getKey ( ) , new CacheStatusResponse ( cacheStatus . getJoinInfo ( ) , cacheStatus . getCurrentTopology ( ) , cacheStatus . getStableTopology ( ) , cacheStatus . getPartitionHandlingManager ( ) . getAvailabilityMode ( ) ) ) ; } } boolean rebalancingEnabled = true ; // Avoid adding a direct dependency to the ClusterTopologyManager ReplicableCommand command = new CacheTopologyControlCommand ( null , CacheTopologyControlCommand . Type . POLICY_GET_STATUS , transport . getAddress ( ) , transport . getViewId ( ) ) ; try { gcr . wireDependencies ( command ) ; SuccessfulResponse response = ( SuccessfulResponse ) command . invoke ( ) ; rebalancingEnabled = ( Boolean ) response . getResponseValue ( ) ; } catch ( Throwable t ) { log . warn ( "Failed to obtain the rebalancing status" , t ) ; } log . debugf ( "Sending cluster status response for view %d" , viewId ) ; return new ManagerStatusResponse ( caches , rebalancingEnabled ) ; }
called by the coordinator