public void sendOutState ( State < Serializable , Serializable > state , String checkpointId , boolean spillState , String location ) { lock . lock ( ) ; try { // flush all the current data before sending the state flushRemaining ( ) ; // serialize the state byte [ ] serializedState = serializer . serialize ( state ) ; CheckpointManager . InstanceStateCheckpoint . Builder instanceStateBuilder = CheckpointManager . InstanceStateCheckpoint . newBuilder ( ) ; instanceStateBuilder . setCheckpointId ( checkpointId ) ; if ( spillState ) { FileUtils . cleanDir ( location ) ; String stateLocation = location + checkpointId + "-" + UUID . randomUUID ( ) ; if ( ! FileUtils . writeToFile ( stateLocation , serializedState , true ) ) { throw new RuntimeException ( "failed to spill state. Bailing out..." ) ; } instanceStateBuilder . setStateLocation ( stateLocation ) ; } else { instanceStateBuilder . setState ( ByteString . copyFrom ( serializedState ) ) ; } CheckpointManager . StoreInstanceStateCheckpoint storeRequest = CheckpointManager . StoreInstanceStateCheckpoint . newBuilder ( ) . setState ( instanceStateBuilder . build ( ) ) . build ( ) ; // Put the checkpoint to out stream queue outQueue . offer ( storeRequest ) ; } finally { lock . unlock ( ) ; } }
Send out the instance s state with corresponding checkpointId . If spillState is True the actual state is spill to disk and only the state location is sent out .