DefaultTransWiring implements Runnable { @Override public void run() { final RowProducer rowProducer; final Trans serviceTrans = dataServiceExecutor.getServiceTrans(); final Trans genTrans = dataServiceExecutor.getGenTrans(); try { rowProducer = dataServiceExecutor.addRowProducer(); } catch ( KettleException e ) { throw Throwables.propagate( e ); } StepInterface serviceStep = serviceTrans.findRunThread( dataServiceExecutor.getService().getStepname() ); if ( serviceStep == null ) { throw Throwables.propagate( new KettleException( "Service step is not accessible" ) ); } serviceStep.addRowListener( new DefaultTransWiringRowAdapter( serviceTrans, genTrans, rowProducer ) ); serviceStep.addStepListener( new StepAdapter() { @Override public void stepFinished( Trans trans, StepMeta stepMeta, StepInterface step ) { rowProducer.finished(); } } ); dataServiceExecutor.getGenTrans() .findRunThread( dataServiceExecutor.getResultStepName() ) .addStepListener( new StepAdapter() { @Override public void stepFinished( Trans trans, StepMeta stepMeta, StepInterface step ) { if ( serviceTrans.isRunning() ) { trans.getLogChannel().logBasic( "Query finished, stopping service transformation" ); serviceTrans.stopAll(); } } } ); } DefaultTransWiring( DataServiceExecutor dataServiceExecutor ); @Override void run();  }
@Test public void testInaccessibleServiceStep() throws KettleException { when( serviceTrans.findRunThread( "step" ) ).thenReturn( null ); try { defaultTransWiring.run(); } catch ( Throwable t ) { assertEquals( "Service step is not accessible", t.getCause().getMessage().trim() ); } }