DataServiceExecutor { public void waitUntilFinished() { if ( !service.isStreaming() ) { serviceTrans.waitUntilFinished(); genTrans.waitUntilFinished(); } else { try { while ( !genTrans.isFinishedOrStopped() && !genTransformationPushBasedIsFinished.get() ) { Thread.sleep( 50 ); } } catch ( InterruptedException e ) { throw new RuntimeException( e ); } } } private  DataServiceExecutor( Builder builder ); static void writeMetadata( DataOutputStream dos, String... metadatas ); DataServiceExecutor executeQuery( final DataOutputStream dos ); DataServiceExecutor executeQuery( Observer<RowMetaAndData> consumer ); DataServiceExecutor executeStreamingQuery( final Observer<List<RowMetaAndData>> streamingConsumer, boolean pollingMode ); String getStreamingGenTransCacheKey(); DataServiceExecutor executeDefaultQuery( final Observer<RowMetaAndData> consumer ); DataServiceExecutor executeQuery(); void executeListeners( ExecutionPoint... stages ); RowProducer addRowProducer(); void waitUntilFinished(); TransMeta getServiceTransMeta(); TransMeta getGenTransMeta(); DataServiceMeta getService(); Trans getServiceTrans(); Trans getGenTrans(); String getServiceName(); Map<String, String> getParameters(); String getId(); Boolean hasErrors(); static String calculateTransname( SQL sql, boolean isService ); void stop( boolean stopTrans ); boolean isStopped(); SQL getSql(); String getResultStepName(); int getRowLimit(); int getServiceRowLimit(); ListMultimap<ExecutionPoint, Runnable> getListenerMap();  }
@Test public void testWaitUntilFinished() throws Exception { SQL sql = new SQL( "SELECT * FROM " + DATA_SERVICE_NAME ); when( serviceTrans.getTransMeta().listParameters() ).thenReturn( new String[ 0 ] ); PushDownOptimizationMeta optimization = mock( PushDownOptimizationMeta.class ); when( optimization.isEnabled() ).thenReturn( true ); dataService.getPushDownOptimizationMeta().add( optimization ); IMetaStore metastore = mock( IMetaStore.class ); DataServiceExecutor executor = new DataServiceExecutor.Builder( sql, dataService, context ). serviceTrans( serviceTrans ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). metastore( metastore ). build(); executor.waitUntilFinished(); verify( genTrans, times( 0 ) ).isFinishedOrStopped(); verify( serviceTrans, times( 1 ) ).waitUntilFinished(); }