@Test public void execute_interrupted() throws KettleException { final long waitTime = 1000 * 10; final List<String> loggedErrors = new ArrayList<String>(); TestJobEntry je = new TestJobEntry( waitTime ) { @Override public void logError( String message, Throwable e ) { loggedErrors.add( message ); } }; final Job parentJob = new Job( "test", null, null ); Thread t = new Thread() { @Override public void run() { try { Thread.sleep( 1000 ); } catch ( InterruptedException e ) { throw new RuntimeException( e ); } parentJob.stopAll(); } }; je.setParentJob( parentJob ); Result result = new Result(); t.start(); long start = System.currentTimeMillis(); je.execute( result, 0 ); long end = System.currentTimeMillis(); assertTrue( "Total runtime should be less than the wait time if we were properly interrupted", ( end - start ) < waitTime ); Assert.assertEquals( 1, result.getNrErrors() ); assertFalse( result.getResult() ); Assert.assertEquals( 1, loggedErrors.size() ); }
@Override public Result execute( Result result, int i ) throws KettleException { if ( !isValid( getJobConfig() ) ) { setJobResultFailed( result ); return result; } final Result jobResult = result; result.setResult( true ); Thread t = new Thread( getExecutionRunnable( jobResult ) ); t.setUncaughtExceptionHandler( new Thread.UncaughtExceptionHandler() { @Override public void uncaughtException( Thread t, Throwable e ) { handleUncaughtThreadException( t, e, jobResult ); } } ); t.start(); if ( JobEntryUtils.asBoolean( getJobConfig().getBlockingExecution(), variables ) ) { while ( !parentJob.isStopped() && t.isAlive() ) { try { t.join( JobEntryUtils.asLong( getJobConfig().getBlockingPollingInterval(), variables ) ); } catch ( InterruptedException ex ) { break; } } if ( t.isAlive() ) { t.interrupt(); setJobResultFailed( result ); } try { t.join( 10 * 1000 ); } catch ( InterruptedException e ) { } } return result; }
AbstractJobEntry extends JobEntryBase implements Cloneable,
    JobEntryInterface { @Override public Result execute( Result result, int i ) throws KettleException { if ( !isValid( getJobConfig() ) ) { setJobResultFailed( result ); return result; } final Result jobResult = result; result.setResult( true ); Thread t = new Thread( getExecutionRunnable( jobResult ) ); t.setUncaughtExceptionHandler( new Thread.UncaughtExceptionHandler() { @Override public void uncaughtException( Thread t, Throwable e ) { handleUncaughtThreadException( t, e, jobResult ); } } ); t.start(); if ( JobEntryUtils.asBoolean( getJobConfig().getBlockingExecution(), variables ) ) { while ( !parentJob.isStopped() && t.isAlive() ) { try { t.join( JobEntryUtils.asLong( getJobConfig().getBlockingPollingInterval(), variables ) ); } catch ( InterruptedException ex ) { break; } } if ( t.isAlive() ) { t.interrupt(); setJobResultFailed( result ); } try { t.join( 10 * 1000 ); } catch ( InterruptedException e ) { } } return result; } }
AbstractJobEntry extends JobEntryBase implements Cloneable,
    JobEntryInterface { @Override public Result execute( Result result, int i ) throws KettleException { if ( !isValid( getJobConfig() ) ) { setJobResultFailed( result ); return result; } final Result jobResult = result; result.setResult( true ); Thread t = new Thread( getExecutionRunnable( jobResult ) ); t.setUncaughtExceptionHandler( new Thread.UncaughtExceptionHandler() { @Override public void uncaughtException( Thread t, Throwable e ) { handleUncaughtThreadException( t, e, jobResult ); } } ); t.start(); if ( JobEntryUtils.asBoolean( getJobConfig().getBlockingExecution(), variables ) ) { while ( !parentJob.isStopped() && t.isAlive() ) { try { t.join( JobEntryUtils.asLong( getJobConfig().getBlockingPollingInterval(), variables ) ); } catch ( InterruptedException ex ) { break; } } if ( t.isAlive() ) { t.interrupt(); setJobResultFailed( result ); } try { t.join( 10 * 1000 ); } catch ( InterruptedException e ) { } } return result; } AbstractJobEntry(); protected  AbstractJobEntry( LogChannelInterface logChannelInterface ); }
AbstractJobEntry extends JobEntryBase implements Cloneable,
    JobEntryInterface { @Override public Result execute( Result result, int i ) throws KettleException { if ( !isValid( getJobConfig() ) ) { setJobResultFailed( result ); return result; } final Result jobResult = result; result.setResult( true ); Thread t = new Thread( getExecutionRunnable( jobResult ) ); t.setUncaughtExceptionHandler( new Thread.UncaughtExceptionHandler() { @Override public void uncaughtException( Thread t, Throwable e ) { handleUncaughtThreadException( t, e, jobResult ); } } ); t.start(); if ( JobEntryUtils.asBoolean( getJobConfig().getBlockingExecution(), variables ) ) { while ( !parentJob.isStopped() && t.isAlive() ) { try { t.join( JobEntryUtils.asLong( getJobConfig().getBlockingPollingInterval(), variables ) ); } catch ( InterruptedException ex ) { break; } } if ( t.isAlive() ) { t.interrupt(); setJobResultFailed( result ); } try { t.join( 10 * 1000 ); } catch ( InterruptedException e ) { } } return result; } AbstractJobEntry(); protected  AbstractJobEntry( LogChannelInterface logChannelInterface ); T getJobConfig(); void setJobConfig( T jobConfig ); @Override boolean evaluates(); @Override boolean isUnconditional(); @Override String getXML(); @Override void loadXML( Node node, List<DatabaseMeta> databaseMetas, List<SlaveServer> slaveServers,
      Repository repository ); @Override void loadRep( Repository rep, ObjectId id_jobentry, List<DatabaseMeta> databases,
      List<SlaveServer> slaveServers ); @Override void saveRep( Repository rep, ObjectId id_job ); @Override Result execute( Result result, int i ); void setJobResultFailed( Result jobResult ); boolean isValid( T config ); VariableSpace getVariableSpace(); abstract List<String> getValidationWarnings( T config ); }
AbstractJobEntry extends JobEntryBase implements Cloneable,
    JobEntryInterface { @Override public Result execute( Result result, int i ) throws KettleException { if ( !isValid( getJobConfig() ) ) { setJobResultFailed( result ); return result; } final Result jobResult = result; result.setResult( true ); Thread t = new Thread( getExecutionRunnable( jobResult ) ); t.setUncaughtExceptionHandler( new Thread.UncaughtExceptionHandler() { @Override public void uncaughtException( Thread t, Throwable e ) { handleUncaughtThreadException( t, e, jobResult ); } } ); t.start(); if ( JobEntryUtils.asBoolean( getJobConfig().getBlockingExecution(), variables ) ) { while ( !parentJob.isStopped() && t.isAlive() ) { try { t.join( JobEntryUtils.asLong( getJobConfig().getBlockingPollingInterval(), variables ) ); } catch ( InterruptedException ex ) { break; } } if ( t.isAlive() ) { t.interrupt(); setJobResultFailed( result ); } try { t.join( 10 * 1000 ); } catch ( InterruptedException e ) { } } return result; } AbstractJobEntry(); protected  AbstractJobEntry( LogChannelInterface logChannelInterface ); T getJobConfig(); void setJobConfig( T jobConfig ); @Override boolean evaluates(); @Override boolean isUnconditional(); @Override String getXML(); @Override void loadXML( Node node, List<DatabaseMeta> databaseMetas, List<SlaveServer> slaveServers,
      Repository repository ); @Override void loadRep( Repository rep, ObjectId id_jobentry, List<DatabaseMeta> databases,
      List<SlaveServer> slaveServers ); @Override void saveRep( Repository rep, ObjectId id_job ); @Override Result execute( Result result, int i ); void setJobResultFailed( Result jobResult ); boolean isValid( T config ); VariableSpace getVariableSpace(); abstract List<String> getValidationWarnings( T config );  }