DataServiceExecutor { @VisibleForTesting protected void wrapupConsumerResources( Observer<RowMetaAndData> consumer ) { consumer.onComplete(); StreamingServiceTransExecutor serviceTransExecutor = context.getServiceTransExecutor( streamServiceKey ); String streamingGenTransCacheKey = getStreamingGenTransCacheKey(); serviceTransExecutor.clearCacheByKey( streamingGenTransCacheKey ); context.removeServiceTransExecutor( serviceTransExecutor.getKey().getDataServiceId() ); genTransformationPushBasedIsFinished.set( true ); } 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 testWrapupConsumerResources() throws KettleException { SQL sql = new SQL( "SELECT * FROM " + DATA_SERVICE_NAME ); when( serviceTransExecutor.getServiceTrans() ).thenReturn( serviceTrans ); when( serviceTransExecutor.getKey() ).thenReturn( key ); when( serviceTransExecutor.getWindowMaxRowLimit() ).thenReturn( 1000L ); when( serviceTransExecutor.getWindowMaxTimeLimit() ).thenReturn( 1000L ); when( sqlTransGenerator.getSql() ).thenReturn( sql ); when( serviceTrans.getTransMeta() ).thenReturn( dataService.getServiceTrans() ); context.addServiceTransExecutor( serviceTransExecutor ); dataService.setStreaming( true ); StreamingGeneratedTransExecution streamWiring = new StreamingGeneratedTransExecution( context.getServiceTransExecutor( key ), genTrans, consumer, pollingMode, sqlTransGenerator.getInjectorStepName(), sqlTransGenerator.getResultStepName(), sqlTransGenerator.getSql().getSqlString(), IDataServiceClientService.StreamingMode.ROW_BASED, 10, 1, 10000, "" ); String streamingGenTransCacheKey = WindowParametersHelper.getCacheKey( sqlTransGenerator.getSql().getSqlString(), IDataServiceClientService.StreamingMode.ROW_BASED, 10, 1, (int) serviceTransExecutor.getWindowMaxRowLimit(), serviceTransExecutor.getWindowMaxTimeLimit(), 10000, serviceTransExecutor.getKey().hashCode() ); context.addStreamingGeneratedTransExecution( streamingGenTransCacheKey, streamWiring ); dataService.setStreaming( true ); IMetaStore metastore = mock( IMetaStore.class ); DataServiceExecutor executor = new DataServiceExecutor.Builder( sql, dataService, context ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). metastore( metastore ). enableMetrics( false ). normalizeConditions( false ). rowLimit( 50 ). windowEvery( 1 ). windowSize( 10 ). windowLimit( 10000 ). windowMode( IDataServiceClientService.StreamingMode.ROW_BASED ). build(); AtomicBoolean onCompleteCalled = new AtomicBoolean( false ); PublishSubject<RowMetaAndData> consumer = PublishSubject.create(); consumer.doOnComplete( () -> onCompleteCalled.set( true ) ).subscribe(); executor.wrapupConsumerResources( consumer ); verify( serviceTransExecutor, times( 1 ) ).clearCacheByKey( anyString() ); assertTrue( onCompleteCalled.get() ); }