private void runForBlockingDataTransfer ( SiteProcedureConnection siteConnection ) { boolean sourcesReady = false ; RestoreWork restoreWork = m_dataSink . poll ( m_snapshotBufferAllocator ) ; if ( restoreWork != null ) { restoreBlock ( restoreWork , siteConnection ) ; sourcesReady = true ; } // The completion monitor may fire even if m_dataSink has not reached EOF in the case that there's no // replicated table in the database, so check for both conditions. if ( m_dataSink . isEOF ( ) || m_snapshotCompletionMonitor . isDone ( ) ) { // No more data from this data sink, close and remove it from the list m_dataSink . close ( ) ; if ( m_streamSnapshotMb != null ) { VoltDB . instance ( ) . getHostMessenger ( ) . removeMailbox ( m_streamSnapshotMb . getHSId ( ) ) ; m_streamSnapshotMb = null ; ELASTICLOG . debug ( m_whoami + " data transfer is finished" ) ; } if ( m_snapshotCompletionMonitor . isDone ( ) ) { try { SnapshotCompletionEvent event = m_snapshotCompletionMonitor . get ( ) ; siteConnection . setDRProtocolVersion ( event . drVersion ) ; assert ( event != null ) ; ELASTICLOG . debug ( "P" + m_partitionId + " noticed data transfer completion" ) ; m_completionAction . setSnapshotTxnId ( event . multipartTxnId ) ; setJoinComplete ( siteConnection , event . exportSequenceNumbers , event . drSequenceNumbers , event . drMixedClusterSizeConsumerState , false /* requireExistingSequenceNumbers */ , event . clusterCreateTime ) ; } catch ( InterruptedException e ) { // isDone() already returned true, this shouldn't happen VoltDB . crashLocalVoltDB ( "Impossible interruption happend" , true , e ) ; } catch ( ExecutionException e ) { VoltDB . crashLocalVoltDB ( "Error waiting for snapshot to finish" , true , e ) ; } } else { m_taskQueue . offer ( this ) ; } } else { // The sources are not set up yet, don't block the site, // return here and retry later. returnToTaskQueue ( sourcesReady ) ; } }
Blocking transfer all partitioned table data and notify the coordinator .