public static void requestSnapshot ( final long clientHandle , final String path , final String nonce , final boolean blocking , final SnapshotFormat format , final SnapshotPathType stype , final String data , final SnapshotResponseHandler handler , final boolean notifyChanges ) { final SnapshotInitiationInfo snapInfo = new SnapshotInitiationInfo ( path , nonce , blocking , format , stype , data ) ; final SimpleClientResponseAdapter adapter = new SimpleClientResponseAdapter ( ClientInterface . SNAPSHOT_UTIL_CID , "SnapshotUtilAdapter" , true ) ; final LinkedBlockingQueue < ClientResponse > responses = new LinkedBlockingQueue < ClientResponse > ( ) ; adapter . registerCallback ( clientHandle , new SimpleClientResponseAdapter . Callback ( ) { @ Override public void handleResponse ( ClientResponse response ) { responses . offer ( response ) ; } } ) ; final SnapshotDaemon sd = VoltDB . instance ( ) . getClientInterface ( ) . getSnapshotDaemon ( ) ; Runnable work = new Runnable ( ) { @ Override public void run ( ) { ClientResponse response = null ; // abort if unable to succeed in 2 hours final long startTime = System . currentTimeMillis ( ) ; boolean hasRequested = false ; while ( System . currentTimeMillis ( ) - startTime <= TimeUnit . HOURS . toMillis ( 2 ) ) { try { if ( ! hasRequested ) { sd . createAndWatchRequestNode ( clientHandle , adapter , snapInfo , notifyChanges ) ; hasRequested = true ; } try { response = responses . poll ( TimeUnit . HOURS . toMillis ( 2 ) - ( System . currentTimeMillis ( ) - startTime ) , TimeUnit . MILLISECONDS ) ; if ( response == null ) { break ; } } catch ( InterruptedException e ) { VoltDB . crashLocalVoltDB ( "Should never happen" , true , e ) ; } VoltTable [ ] results = response . getResults ( ) ; if ( response . getStatus ( ) != ClientResponse . SUCCESS ) { break ; } else if ( isSnapshotInProgress ( results ) ) { // retry after a second Thread . sleep ( 1000 ) ; // Request again hasRequested = false ; continue ; } else if ( isSnapshotQueued ( results ) && notifyChanges ) { //Wait for an update on the queued state via ZK continue ; } else { // other errors are not recoverable break ; } } catch ( ForwardClientException e ) { //This happens when something goes wrong in the snapshot daemon //I think it will always be that there was an existing snapshot request //It should eventually terminate and then we can submit one. try { Thread . sleep ( 5000 ) ; } catch ( InterruptedException e1 ) { } new VoltLogger ( "SNAPSHOT" ) . warn ( "Partition detection is unable to submit a snapshot request " + "because one already exists. Retrying." ) ; continue ; } catch ( InterruptedException ignore ) { } } handler . handleResponse ( response ) ; } } ; // Use an executor service here to avoid explosion of threads??? ThreadFactory factory = CoreUtils . getThreadFactory ( "Snapshot Request - " + nonce ) ; Thread workThread = factory . newThread ( work ) ; workThread . start ( ) ; }
Request a new snapshot . It will retry for a couple of times . If it doesn t succeed in the specified time an error response will be sent to the response handler otherwise a success response will be passed to the handler .