private void commitCommand ( CommandRequest request , CompletableFuture < CommandResponse > future ) { final long term = raft . getTerm ( ) ; final long timestamp = System . currentTimeMillis ( ) ; CommandEntry command = new CommandEntry ( term , timestamp , request . session ( ) , request . sequenceNumber ( ) , request . operation ( ) ) ; appendAndCompact ( command ) . whenCompleteAsync ( ( entry , error ) -> { if ( error != null ) { Throwable cause = Throwables . getRootCause ( error ) ; if ( Throwables . getRootCause ( error ) instanceof StorageException . TooLarge ) { log . warn ( "Failed to append command {}" , command , cause ) ; future . complete ( CommandResponse . builder ( ) . withStatus ( RaftResponse . Status . ERROR ) . withError ( RaftError . Type . PROTOCOL_ERROR ) . build ( ) ) ; } else { future . complete ( CommandResponse . builder ( ) . withStatus ( RaftResponse . Status . ERROR ) . withError ( RaftError . Type . COMMAND_FAILURE ) . build ( ) ) ; } return ; } // Replicate the command to followers. appender . appendEntries ( entry . index ( ) ) . whenComplete ( ( commitIndex , commitError ) -> { raft . checkThread ( ) ; if ( isRunning ( ) ) { // If the command was successfully committed, apply it to the state machine. if ( commitError == null ) { raft . getServiceManager ( ) . < OperationResult > apply ( entry . index ( ) ) . whenComplete ( ( r , e ) -> { completeOperation ( r , CommandResponse . builder ( ) , e , future ) ; } ) ; } else { future . complete ( CommandResponse . builder ( ) . withStatus ( RaftResponse . Status . ERROR ) . withError ( RaftError . Type . COMMAND_FAILURE ) . build ( ) ) ; } } else { future . complete ( CommandResponse . builder ( ) . withStatus ( RaftResponse . Status . ERROR ) . withError ( RaftError . Type . COMMAND_FAILURE ) . build ( ) ) ; } } ) ; } , raft . getThreadContext ( ) ) ; }
Commits a command .