private void addCallback ( int rc , LedgerHandle handle , long entryId , Object ctx ) { Write write = ( Write ) ctx ; try { assert handle . getId ( ) == write . getWriteLedger ( ) . ledger . getId ( ) : "Handle.Id mismatch: " + write . getWriteLedger ( ) . ledger . getId ( ) + " vs " + handle . getId ( ) ; write . setEntryId ( entryId ) ; if ( rc == 0 ) { // Successful write. If we get this, then by virtue of how the Writes are executed (always wait for writes // in previous ledgers to complete before initiating, and BookKeeper guaranteeing that all writes in this // ledger prior to this writes are done), it is safe to complete the callback future now. completeWrite ( write ) ; return ; } // Convert the response code into an Exception. Eventually this will be picked up by the WriteProcessor which // will retry it or fail it permanently (this includes exceptions from rollovers). handleWriteException ( rc , write ) ; } catch ( Throwable ex ) { // Most likely a bug in our code. We still need to fail the write so we don't leave it hanging. write . fail ( ex , ! isRetryable ( ex ) ) ; } finally { // Process all the appends in the queue after any change. This finalizes the completion, does retries (if needed) // and triggers more appends. try { this . writeProcessor . runAsync ( ) ; } catch ( ObjectClosedException ex ) { // In case of failures, the WriteProcessor may already be closed. We don't want the exception to propagate // to BookKeeper. log . warn ( "{}: Not running WriteProcessor as part of callback due to BookKeeperLog being closed." , this . traceObjectId , ex ) ; } } }
Callback for BookKeeper appends .