private void flushQueues ( ) { // Terminate data loading thread. LOGGER . debug ( "Flush Queues" ) ; try { _queueData . put ( new byte [ 0 ] ) ; _thread . join ( 10000 ) ; if ( _thread . isAlive ( ) ) { _thread . interrupt ( ) ; } } catch ( Exception ex ) { String msg = "Failed to join StreamLoader queue: " + ex . getMessage ( ) ; LOGGER . error ( msg , ex ) ; throw new DataError ( msg , Utils . getCause ( ex ) ) ; } // Put last stage on queue terminate ( ) ; // wait for the processing to finish _put . join ( ) ; _process . join ( ) ; if ( _aborted . get ( ) ) { // Loader was aborted due to an exception. // It was rolled back at that time. //LOGGER.log(Level.WARNING, // "Loader had been previously aborted by error", _abortCause); throw _abortCause ; } }
Flushes data by joining PUT and PROCESS queues