private boolean processCommand ( DatanodeCommand [ ] cmds , long processStartTime ) throws InterruptedException { if ( cmds != null ) { // at each heartbeat the standby offer service will talk to ZK! boolean switchedFromStandbyToPrimary = checkFailover ( ) ; for ( DatanodeCommand cmd : cmds ) { try { // The datanode has received a register command after the failover, this // means that the offerservice thread for the datanode was down for a // while and it most probably did not clean up its deletion queue, hence // force a cleanup. if ( switchedFromStandbyToPrimary && cmd . getAction ( ) == DatanodeProtocol . DNA_REGISTER ) { this . clearPrimary ( ) ; } // The standby service thread is allowed to process only a small set // of valid commands. if ( ! isPrimaryServiceCached ( ) && ! isValidStandbyCommand ( cmd ) ) { LOG . warn ( "Received an invalid command " + cmd . getAction ( ) + " from standby " + this . namenodeAddress ) ; continue ; } if ( processCommand ( cmd , processStartTime ) == false ) { return false ; } } catch ( IOException ioe ) { LOG . warn ( "Error processing datanode Command" , ioe ) ; } } } return true ; }
Process an array of datanode commands . This function has logic to check for failover . Any commands should be processed using this function as an entry point .