DataServiceExecutor { protected void prepareExecution() throws KettleException { ImmutableMultimap.Builder<ExecutionPoint, Runnable> builder = ImmutableMultimap.builder(); builder.putAll( ExecutionPoint.PREPARE, new CopyParameters( parameters, serviceTrans ), new PrepareExecution( serviceTrans ), new PrepareExecution( genTrans ) ); builder.putAll( ExecutionPoint.READY, new DefaultTransWiring( this ) ); builder.putAll( ExecutionPoint.START, new TransStarter( genTrans ), new TransStarter( serviceTrans ) ); listenerMap.putAll( builder.build() ); } 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 testPrepareExecution() throws Exception { SQL sql = new SQL( "SELECT * FROM " + DATA_SERVICE_NAME ); when( serviceTransExecutor.getServiceTrans() ).thenReturn( serviceTrans ); IMetaStore metastore = mock( IMetaStore.class ); Map<String, String> parameters = new HashMap<>( ); parameters.put( "param1", "value1" ); DataServiceExecutor executor = new DataServiceExecutor.Builder( sql, dataService, context ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). metastore( metastore ). enableMetrics( false ). normalizeConditions( false ). parameters( parameters ). build(); assertNotNull( executor.getListenerMap() ); assertNotNull( executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.PREPARE ) ); boolean existsCopyParameters = false; for ( Runnable runnable : executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.PREPARE ) ) { if ( runnable instanceof CopyParameters ) { existsCopyParameters = true; } } assertTrue( existsCopyParameters ); assertEquals( 3, executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.PREPARE ).size() ); assertEquals( 1, executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.READY ).size() ); assertEquals( 2, executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.START ).size() ); assertEquals( 0, executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.OPTIMIZE ).size() ); executor = new DataServiceExecutor.Builder( sql, dataService, context ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). metastore( metastore ). enableMetrics( false ). normalizeConditions( false ). build(); assertNotNull( executor.getListenerMap() ); assertNotNull( executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.PREPARE ) ); existsCopyParameters = false; for ( Runnable runnable : executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.PREPARE ) ) { if ( runnable instanceof CopyParameters ) { existsCopyParameters = true; } } assertTrue( existsCopyParameters ); assertEquals( 3, executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.PREPARE ).size() ); assertEquals( 1, executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.READY ).size() ); assertEquals( 2, executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.START ).size() ); assertEquals( 0, executor.getListenerMap().get( DataServiceExecutor.ExecutionPoint.OPTIMIZE ).size() ); }