DataServiceExecutor { public DataServiceExecutor executeQuery( final DataOutputStream dos ) throws IOException { writeMetadata( dos, getServiceName(), calculateTransname( getSql(), true ), getServiceTrans().getContainerObjectId(), calculateTransname( getSql(), false ), getGenTrans().getContainerObjectId() ); final AtomicBoolean rowMetaWritten = new AtomicBoolean( false ); genTransformationPushBasedIsFinished.set( false ); if ( transListenerFinishTransAdded.compareAndSet( false, true ) ) { getGenTrans().addTransListener( new TransAdapter() { @Override public void transFinished( Trans trans ) throws KettleException { writeMeta( trans, dos, rowMetaWritten ); } } ); } final PublishSubject<RowMetaAndData> consumer = PublishSubject.create(); Disposable[] disposableWrapper = new Disposable[ 1 ]; consumer .doOnError( t -> { logger.error( "Error consuming data from the generated transformation into the consumer", t ); wrapupConsumerResources( consumer ); } ) .doOnComplete( () -> { writeMeta( getGenTrans(), dos, rowMetaWritten ); genTransformationPushBasedIsFinished.set( true ); if ( disposableWrapper[ 0 ] != null ) { disposableWrapper[ 0 ].dispose(); } } ) .doOnSubscribe( disposable -> disposableWrapper[ 0 ] = disposable ) .subscribe( rowMetaAndData -> { try { RowMetaInterface rowMetaInterface = rowMetaAndData.getRowMeta(); writeMeta( rowMetaInterface, dos, rowMetaWritten ); rowMetaInterface.writeData( dos, rowMetaAndData.getData() ); } catch ( Exception e ) { if ( !getServiceTrans().isStopped() ) { throw new KettleStepException( e ); } } } ); return executeQuery( consumer ); } 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 testExecuteQuery() throws Exception { testExecuteQueryAux( true, false ); }