@ Override public void heartbeat ( ) { // Prepare metadata for the next heartbeat BlockHeartbeatReport blockReport = mBlockWorker . getReport ( ) ; BlockStoreMeta storeMeta = mBlockWorker . getStoreMeta ( ) ; // Send the heartbeat and execute the response Command cmdFromMaster = null ; List < alluxio . grpc . Metric > metrics = new ArrayList <> ( ) ; for ( Metric metric : MetricsSystem . allWorkerMetrics ( ) ) { metrics . add ( metric . toProto ( ) ) ; } try { cmdFromMaster = mMasterClient . heartbeat ( mWorkerId . get ( ) , storeMeta . getCapacityBytesOnTiers ( ) , storeMeta . getUsedBytesOnTiers ( ) , blockReport . getRemovedBlocks ( ) , blockReport . getAddedBlocks ( ) , blockReport . getLostStorage ( ) , metrics ) ; handleMasterCommand ( cmdFromMaster ) ; mLastSuccessfulHeartbeatMs = System . currentTimeMillis ( ) ; } catch ( IOException | ConnectionFailedException e ) { // An error occurred, log and ignore it or error if heartbeat timeout is reached if ( cmdFromMaster == null ) { LOG . error ( "Failed to receive master heartbeat command." , e ) ; } else { LOG . error ( "Failed to receive or execute master heartbeat command: {}" , cmdFromMaster . toString ( ) , e ) ; } mMasterClient . disconnect ( ) ; if ( mHeartbeatTimeoutMs > 0 ) { if ( System . currentTimeMillis ( ) - mLastSuccessfulHeartbeatMs >= mHeartbeatTimeoutMs ) { if ( ServerConfiguration . getBoolean ( PropertyKey . TEST_MODE ) ) { throw new RuntimeException ( "Master heartbeat timeout exceeded: " + mHeartbeatTimeoutMs ) ; } // TODO(andrew): Propagate the exception to the main thread and exit there. ProcessUtils . fatalError ( LOG , "Master heartbeat timeout exceeded: %d" , mHeartbeatTimeoutMs ) ; } } } }
Heartbeats to the master node about the change in the worker s managed space .