@ SuppressWarnings ( { "unchecked" , "rawtypes" , "unused" } ) private void commitToVoldemort ( List < String > storeNamesToCommit ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Trying to commit to Voldemort" ) ; } boolean hasError = false ; if ( nodesToStream == null || nodesToStream . size ( ) == 0 ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "No nodes to stream to. Returning." ) ; } return ; } for ( Node node : nodesToStream ) { for ( String store : storeNamesToCommit ) { if ( ! nodeIdStoreInitialized . get ( new Pair ( store , node . getId ( ) ) ) ) continue ; nodeIdStoreInitialized . put ( new Pair ( store , node . getId ( ) ) , false ) ; DataOutputStream outputStream = nodeIdStoreToOutputStreamRequest . get ( new Pair ( store , node . getId ( ) ) ) ; try { ProtoUtils . writeEndOfStream ( outputStream ) ; outputStream . flush ( ) ; DataInputStream inputStream = nodeIdStoreToInputStreamRequest . get ( new Pair ( store , node . getId ( ) ) ) ; VAdminProto . UpdatePartitionEntriesResponse . Builder updateResponse = ProtoUtils . readToBuilder ( inputStream , VAdminProto . UpdatePartitionEntriesResponse . newBuilder ( ) ) ; if ( updateResponse . hasError ( ) ) { hasError = true ; } } catch ( IOException e ) { logger . error ( "Exception during commit" , e ) ; hasError = true ; if ( ! faultyNodes . contains ( node . getId ( ) ) ) faultyNodes . add ( node . getId ( ) ) ; } } } if ( streamingresults == null ) { logger . warn ( "StreamingSession may not have been initialized since Variable streamingresults is null. Skipping callback " ) ; return ; } // remove redundant callbacks if ( hasError ) { logger . info ( "Invoking the Recovery Callback" ) ; Future future = streamingresults . submit ( recoveryCallback ) ; try { future . get ( ) ; } catch ( InterruptedException e1 ) { MARKED_BAD = true ; logger . error ( "Recovery Callback failed" , e1 ) ; throw new VoldemortException ( "Recovery Callback failed" ) ; } catch ( ExecutionException e1 ) { MARKED_BAD = true ; logger . error ( "Recovery Callback failed during execution" , e1 ) ; throw new VoldemortException ( "Recovery Callback failed during execution" ) ; } } else { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Commit successful" ) ; logger . debug ( "calling checkpoint callback" ) ; } Future future = streamingresults . submit ( checkpointCallback ) ; try { future . get ( ) ; } catch ( InterruptedException e1 ) { logger . warn ( "Checkpoint callback failed!" , e1 ) ; } catch ( ExecutionException e1 ) { logger . warn ( "Checkpoint callback failed during execution!" , e1 ) ; } } }
Flush the network buffer and write all entries to the serve . then wait for an ack from the server . This is a blocking call . It is invoked on every Commit batch size of entries It is also called on the close session call