public void stopStep( StepMetaDataCombi combi, boolean safeStop ) {
  StepInterface rt = combi.step;
  rt.setStopped( true );
  rt.setSafeStopped( safeStop );
  rt.resumeRunning();

  try {
    rt.stopRunning( combi.meta, combi.data );
  } catch ( Exception e ) {
    log.logError( "Something went wrong while trying to safe stop the transformation: ", e );
  }
  combi.data.setStatus( StepExecutionStatus.STATUS_STOPPED );
  if ( safeStop ) {
    rt.setOutputDone();
  }
}
 

/**
 * This method runs transformations related to PDI-13545.<br/> The scenario is the following: there are two step
 * generating data, the latter of which is a Mapping step. They are followed with a Join Rows step, that has two
 * copies. The last in a row is a Dummy step, named "Last". Since both generating steps output 3 rows ([10, 20, 30]
 * and [1, 2, 3] respectively), the last step must obtain 3*3=9 rows.
 *
 * @param transPath a path to transformation file
 * @throws Exception
 */
private void runTransWhenMappingsIsFollowedByCopiedStep( String transPath ) throws Exception {
  KettleEnvironment.init();

  TransMeta transMeta = new TransMeta( transPath );
  transMeta.setTransformationType( TransMeta.TransformationType.Normal );

  Trans trans = new Trans( transMeta );
  trans.prepareExecution( null );
  trans.startThreads();
  trans.waitUntilFinished();

  assertEquals( 0, trans.getErrors() );

  List<StepInterface> list = trans.findBaseSteps( "Last" );
  assertEquals( 1, list.size() );
  assertEquals( 9, list.get( 0 ).getLinesRead() );
}
 

/**
 * Gets the mapping inputs for each step in the transformation.
 *
 * @return an array of MappingInputs
 */
public MappingInput[] findMappingInput() {
  if ( steps == null ) {
    return null;
  }

  List<MappingInput> list = new ArrayList<>();

  // Look in threads and find the MappingInput step thread...
  for ( int i = 0; i < steps.size(); i++ ) {
    StepMetaDataCombi smdc = steps.get( i );
    StepInterface step = smdc.step;
    if ( step.getStepID().equalsIgnoreCase( "MappingInput" ) ) {
      list.add( (MappingInput) step );
    }
  }
  return list.toArray( new MappingInput[ list.size() ] );
}
 

protected void processRows( StepInterface step, final int maxCalls ) throws Exception {
  for ( int outRowIdx = 0; outRowIdx < maxCalls; outRowIdx++ ) {
    if ( !step.processRow( helper.processRowsStepMetaInterface, helper.processRowsStepDataInterface ) ) {
      break;
    }
  }
}
 

public static double getProcessCount( Context actualContext, Scriptable actualObject, Object[] ArgList,
  Function FunctionContext ) {

  if ( ArgList.length == 1 ) {
    try {
      Object scmO = actualObject.get( "_step_", actualObject );
      StepInterface scm = (StepInterface) Context.jsToJava( scmO, StepInterface.class );
      String strType = Context.toString( ArgList[0] ).toLowerCase();

      if ( strType.equals( "i" ) ) {
        return scm.getLinesInput();
      } else if ( strType.equals( "o" ) ) {
        return scm.getLinesOutput();
      } else if ( strType.equals( "r" ) ) {
        return scm.getLinesRead();
      } else if ( strType.equals( "u" ) ) {
        return scm.getLinesUpdated();
      } else if ( strType.equals( "w" ) ) {
        return scm.getLinesWritten();
      } else if ( strType.equals( "e" ) ) {
        return scm.getLinesRejected();
      } else {
        return 0;
      }
    } catch ( Exception e ) {
      // throw Context.reportRuntimeError(e.toString());
      return 0;
    }
  } else {
    throw Context.reportRuntimeError( "The function call getProcessCount requires 1 argument." );
  }
}
 

public StepInterface getStep(StepMeta stepMeta,
    StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
    Trans trans) {

  return new CassandraInput(stepMeta, stepDataInterface, copyNr, transMeta,
      trans);
}
 

@Test
@PrepareForTest( { Encode.class } )
public void testSniffStepServletEscapesHtmlWhenTransFound() throws ServletException, IOException {
  KettleLogStore.init();
  HttpServletRequest mockHttpServletRequest = mock( HttpServletRequest.class );
  HttpServletResponse mockHttpServletResponse = mock( HttpServletResponse.class );
  Trans mockTrans = mock( Trans.class );
  TransMeta mockTransMeta = mock( TransMeta.class );
  StepInterface mockStepInterface = mock( StepInterface.class );
  List<StepInterface> stepInterfaces = new ArrayList<StepInterface>();
  stepInterfaces.add( mockStepInterface );
  LogChannelInterface mockChannelInterface = mock( LogChannelInterface.class );
  StringWriter out = new StringWriter();
  PrintWriter printWriter = new PrintWriter( out );

  PowerMockito.spy( Encode.class );
  when( mockHttpServletRequest.getContextPath() ).thenReturn( SniffStepServlet.CONTEXT_PATH );
  when( mockHttpServletRequest.getParameter( anyString() ) ).thenReturn( ServletTestUtils.BAD_STRING_TO_TEST );
  when( mockHttpServletResponse.getWriter() ).thenReturn( printWriter );
  when( mockTransformationMap.getTransformation( any( CarteObjectEntry.class ) ) ).thenReturn( mockTrans );
  when( mockTrans.getLogChannel() ).thenReturn( mockChannelInterface );
  when( mockTrans.getLogChannelId() ).thenReturn( "test" );
  when( mockTrans.getTransMeta() ).thenReturn( mockTransMeta );
  when( mockTransMeta.getMaximum() ).thenReturn( new Point( 10, 10 ) );
  when( mockTrans.findBaseSteps( ServletTestUtils.BAD_STRING_TO_TEST ) ).thenReturn( stepInterfaces );

  sniffStepServlet.doGet( mockHttpServletRequest, mockHttpServletResponse );
  assertFalse( ServletTestUtils.hasBadText( ServletTestUtils.getInsideOfTag( "H1", out.toString() ) ) );

  PowerMockito.verifyStatic( atLeastOnce() );
  Encode.forHtml( anyString() );
}
 

@Test( expected = KettleException.class )
public void testVerifyTransMetaOutKeyNotDefined() throws KettleException {
  String inputStepName = "inputStepName";
  String outputStepName = "outputStepName";
  StepMeta inputStepMeta = mock( StepMeta.class );
  when( transMeta.findStep( inputStepName ) ).thenReturn( inputStepMeta );
  RowMetaInterface rowMetaInterface = mock( RowMetaInterface.class );
  when( rowMetaInterface.getFieldNames() ).thenReturn( new String[] { "key", "value" } );
  when( transMeta.getStepFields( inputStepMeta ) ).thenReturn( rowMetaInterface );
  Trans trans = mock( Trans.class );
  when( transFactory.create( transMeta ) ).thenReturn( trans );
  when( trans.getStepInterface( inputStepName, 0 ) ).thenReturn( mock( StepInterface.class ) );
  final StepMeta outputStepMeta = mock( StepMeta.class );
  StepMetaInterface stepMetaInterface = mock( StepMetaInterface.class );
  when( outputStepMeta.getStepMetaInterface() ).thenReturn( stepMetaInterface );
  when( transMeta.findStep( outputStepName ) ).thenReturn( outputStepMeta );
  RowMetaInterface outputRowMetaInterface = mock( RowMetaInterface.class );
  when( transMeta.getStepFields( outputStepMeta ) ).thenReturn( outputRowMetaInterface );
  when( outputRowMetaInterface.getFieldNames() ).thenReturn( new String[] {} );
  try {
    pentahoMapReduceJobBuilder.verifyTransMeta( transMeta, inputStepName, outputStepName );
  } catch ( KettleException e ) {
    assertEquals( BaseMessages.getString( PentahoMapReduceJobBuilderImpl.PKG,
      PentahoMapReduceJobBuilderImpl.PENTAHO_MAP_REDUCE_JOB_BUILDER_IMPL_NO_OUTPUT_KEY_ORDINAL, outputStepName ),
      e.getMessage().trim() );
    throw e;
  }
}
 

/**
 * Create, execute, and return the row listener attached to the output step with complete results from the execution.
 *
 * @param checkSumType
 *          Type of checksum to use (the array index of {@link CheckSumMeta#checksumtypeCodes})
 * @param compatibilityMode
 *          Use compatibility mode for CheckSum
 * @param fieldSeparatorString
 *          The string separate multiple fields with
 * @param inputs
 *          Array of objects representing row data
 * @param inputValueMetas
 *          metas to be processed
 * @return RowListener with results.
 */
private MockRowListener executeHexTest( int checkSumType, boolean compatibilityMode, boolean oldChecksumBehaviour,
    String fieldSeparatorString, Object[] inputs, ValueMetaInterface... inputValueMetas ) throws Exception {

  String[] fieldNames = new String[inputValueMetas.length];
  RowMeta inputRowMeta = new RowMeta();
  for ( int i = 0; i < inputValueMetas.length; i++ ) {
    inputRowMeta.addValueMeta( inputValueMetas[i] );
    fieldNames[i] = inputValueMetas[i].getName();
  }

  Trans trans =
      buildHexadecimalChecksumTrans( checkSumType, compatibilityMode, oldChecksumBehaviour, fieldSeparatorString,
          fieldNames );

  trans.prepareExecution( null );

  StepInterface output = trans.getRunThread( "Output", 0 );
  MockRowListener listener = new MockRowListener();
  output.addRowListener( listener );

  RowProducer rp = trans.addRowProducer( "CheckSum", 0 );

  ( (BaseStep) trans.getRunThread( "CheckSum", 0 ) ).setInputRowMeta( inputRowMeta );

  trans.startThreads();

  rp.putRow( inputRowMeta, inputs );
  rp.finished();

  trans.waitUntilFinished();
  trans.stopAll();
  trans.cleanup();
  return listener;
}
 

@Test
public void testStepShouldStopProcessingInput_IfUnderlyingTransitionIsStopped() throws Exception {

  MappingInput mappingInput = mock( MappingInput.class );
  when( mappingInput.getStepname() ).thenReturn( MAPPING_INPUT_STEP_NAME );
  stepMockHelper.processRowsStepDataInterface.mappingInput = mappingInput;

  RowProducer rowProducer = mock( RowProducer.class );
  when( rowProducer.putRow( any( RowMetaInterface.class ), any( Object[].class ), anyBoolean() ) )
    .thenReturn( true );

  StepInterface stepInterface = mock( StepInterface.class );

  Trans mappingTrans = mock( Trans.class );
  when( mappingTrans.addRowProducer( anyString(), anyInt() ) ).thenReturn( rowProducer );
  when( mappingTrans.findStepInterface( anyString(), anyInt() ) ).thenReturn( stepInterface );
  when( mappingTrans.isFinishedOrStopped() ).thenReturn( Boolean.FALSE ).thenReturn( Boolean.TRUE );
  stepMockHelper.processRowsStepDataInterface.mappingTrans = mappingTrans;

  MappingOutput mappingOutput = mock( MappingOutput.class );
  when( mappingOutput.getStepname() ).thenReturn( MAPPING_OUTPUT_STEP_NAME );
  stepMockHelper.processRowsStepDataInterface.mappingOutput = mappingOutput;


  smp = new SimpleMapping( stepMockHelper.stepMeta, stepMockHelper.stepDataInterface, 0, stepMockHelper.transMeta,
    stepMockHelper.trans );
  smp.init( stepMockHelper.initStepMetaInterface, simpleMpData );
  smp.addRowSetToInputRowSets( stepMockHelper.getMockInputRowSet( new Object[] { } ) );
  smp.addRowSetToInputRowSets( stepMockHelper.getMockInputRowSet( new Object[] { } ) );

  assertTrue(
    smp.processRow( stepMockHelper.processRowsStepMetaInterface, stepMockHelper.processRowsStepDataInterface ) );
  assertFalse(
    smp.processRow( stepMockHelper.processRowsStepMetaInterface, stepMockHelper.processRowsStepDataInterface ) );

}
 

private void processRow( TransProcessControl control ) throws Exception {
  Trans trans = new Trans( transMeta );
  trans.prepareExecution( null );

  RowProducer rp = trans.addRowProducer( injectorStepname, 0 );
  trans.startThreads();
  generateData( rp );
  rp.finished();
  StepInterface si = trans.getStepInterface( synchronizeAfterMergeStepname, 0 );
  switch ( control ) {
    case ITTERUPT:
      trans.stopAll();
      while (  !si.getStatus().equals( StepExecutionStatus.STATUS_STOPPED ) ) {
        //wait until transformation does not stopped
      };
      break;
    case WAIT:
    default:
      trans.waitUntilFinished();
      assertEquals( "Step still started", StepExecutionStatus.STATUS_FINISHED, si.getStatus() );
      break;
  }
  assertEquals( "Unexpected error occurred",  0, si.getErrors() );

  Field field = SynchronizeAfterMerge.class.getDeclaredField( "data" );
  field.setAccessible( true );
  SynchronizeAfterMergeData  data = (SynchronizeAfterMergeData) field.get( si );
  //should be closed and set null after finish transformation
  assertNull( data.db.getConnection() );
}
 

/**
 * This checks transformation initialization when using many to many copies.
 * 
 * @throws KettleException
 */
@Test
public void testManyToManyCopies() throws KettleException {
  prepareStepMetas_x2_x2();

  trans.prepareExecution( new String[] {} );
  List<RowSet> rowsets = trans.getRowsets();
  assertTrue( !rowsets.isEmpty() );
  assertEquals( "We have 2 rowsets finally", 2, rowsets.size() );
  assertEquals( "We have 4 steps: 2 copies of producer and 2 copies of consumer", 4, trans.getSteps().size() );

  // Ok, examine initialized steps now.
  StepInterface stepOne0 = getStepByName( S10 );
  assertTrue( "1 step have no input row sets", stepOne0.getInputRowSets().isEmpty() );
  assertEquals( "1 step have 1 output rowsets", 1, stepOne0.getOutputRowSets().size() );

  StepInterface stepOne1 = getStepByName( S11 );
  assertTrue( "1 step have no input row sets", stepOne1.getInputRowSets().isEmpty() );
  assertEquals( "1 step have 1 output rowsets", 1, stepOne1.getOutputRowSets().size() );

  StepInterface stepTwo0 = getStepByName( S20 );
  Assert.assertEquals( "2.0 step have 1 input row sets", 1, stepTwo0.getInputRowSets().size() );
  Assert.assertTrue( "2.0 step have no output row sets", stepTwo0.getOutputRowSets().isEmpty() );

  StepInterface stepTwo1 = getStepByName( S21 );
  Assert.assertEquals( "2.1 step have 1 input row sets", 1, stepTwo1.getInputRowSets().size() );
  Assert.assertTrue( "2.1 step have no output row sets", stepTwo1.getOutputRowSets().isEmpty() );
}
 

@Before
public void setUp() throws Exception {
  stepInterface = mock( StepInterface.class );
  rowSet = mock( RowSet.class );
  rowProducer = new RowProducer( stepInterface, rowSet );
  rowMeta = mock( RowMetaInterface.class );
  rowData = new Object[]{};
}
 

/**
 * Find the executing step copy for the step with the specified name and copy number
 *
 * @param stepname the step name
 * @param copynr
 * @return the executing step found or null if no copy could be found.
 */
public StepInterface findStepInterface( String stepname, int copyNr ) {
  if ( steps == null ) {
    return null;
  }

  for ( int i = 0; i < steps.size(); i++ ) {
    StepMetaDataCombi sid = steps.get( i );
    StepInterface stepInterface = sid.step;
    if ( stepInterface.getStepname().equalsIgnoreCase( stepname ) && sid.copy == copyNr ) {
      return stepInterface;
    }
  }
  return null;
}
 

/**
 * Find the run thread for the step with the specified name.
 *
 * @param stepname the step name
 * @return a StepInterface object corresponding to the run thread for the specified step
 */
public StepInterface findRunThread( String stepname ) {
  if ( steps == null ) {
    return null;
  }

  for ( int i = 0; i < steps.size(); i++ ) {
    StepMetaDataCombi sid = steps.get( i );
    StepInterface step = sid.step;
    if ( step.getStepname().equalsIgnoreCase( stepname ) ) {
      return step;
    }
  }
  return null;
}
 

/**
 * Gets the run thread for the step at the specified index.
 *
 * @param i the index of the desired step
 * @return a StepInterface object corresponding to the run thread for the specified step
 */
public StepInterface getRunThread( int i ) {
  if ( steps == null ) {
    return null;
  }
  return steps.get( i ).step;
}
 

private void subscribeToOpLogging() throws KettleException {
  transformation.getOperations().stream().forEach( operation -> {
    try {
      messageEventService.addHandler( Util.getOperationLogEvent( operation.getId() ),
        new MessageEventHandler() {
          @Override
          public void execute( Message message ) throws MessageEventHandlerExecutionException {
            PDIEvent<RemoteSource, LogEntry> event = (PDIEvent<RemoteSource, LogEntry>) message;
            LogEntry logEntry = event.getData();
            StepInterface stepInterface = findStepInterface( operation.getId(), 0 );
            if ( stepInterface != null ) {
              // This is intended to put a red error (dash) on the step in PDI.
              // In order to do that 3 things are needed: errors have to be set
              // to a positive number, the state is stopped state (not finished)
              // and Error log on the step (done just below this if statement)
              if ( LogLevel.ERROR.equals( logEntry.getLogLogLevel() ) ) {
                stepInterface.setErrors( 1 );
                stepInterface.setStopped( true );
              }

              LogChannelInterface logChannel = stepInterface.getLogChannel();
              logToChannel( logChannel, logEntry );
            } else {
              // Could not find step, log at transformation level instead
              logToChannel( getLogChannel(), logEntry );
            }
          }

          @Override
          public String getIdentifier() {
            return OPERATION_LOG + operation.getKey();
          }
        } );
    } catch ( HandlerRegistrationException e ) {
      getLogChannel().logError( "Error registering message handlers", e );
    }
  } );
}
 

/**
 * Test 'Swim lines partitioning'
 * 
 * @throws KettleException
 */
@Test
public void testSwimLanesPartitioning() throws KettleException {
  prepareStepMetas_cl1_cl1();

  trans.prepareExecution( new String[] {} );
  List<RowSet> rowsets = trans.getRowsets();
  assertTrue( !rowsets.isEmpty() );
  assertEquals( "We have 2 rowsets finally", 2, rowsets.size() );
  assertEquals( "We have 3 steps: 1 producer and 2 copies of consumer since it is partitioned", 4, trans.getSteps()
      .size() );

  // Ok, examine initialized steps now.
  StepInterface stepOne0 = getStepByName( SP10 );
  assertTrue( "1.0 step have no input row sets", stepOne0.getInputRowSets().isEmpty() );
  assertEquals( "1.0 step have 1 output rowsets", 1, stepOne0.getOutputRowSets().size() );

  StepInterface stepOne1 = getStepByName( SP11 );
  assertTrue( "1.1 step have no input row sets", stepOne1.getInputRowSets().isEmpty() );
  assertEquals( "1.1 step have 1 output rowsets", 1, stepOne1.getOutputRowSets().size() );

  StepInterface stepTwo0 = getStepByName( SP20 );
  assertEquals( "2.0 step have 2 input row sets", 1, stepTwo0.getInputRowSets().size() );
  assertTrue( "2.0 step have no output rowsets", stepTwo0.getOutputRowSets().isEmpty() );

  StepInterface stepTwo2 = getStepByName( SP21 );
  assertTrue( "2.2 step have no output row sets", stepTwo2.getOutputRowSets().isEmpty() );
  assertEquals( "2.2 step have 2 output rowsets", 1, stepTwo2.getInputRowSets().size() );
}
 

@Override
public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr, TransMeta transMeta,
    Trans trans ) {
  return new GetVariable( stepMeta, stepDataInterface, cnr, transMeta, trans );
}
 

@Override
public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr,
  TransMeta transMeta, Trans trans ) {
  return new DataGrid( stepMeta, stepDataInterface, cnr, transMeta, trans );
}
 

public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr,
  TransMeta transMeta, Trans trans ) {
  return new RowsFromResult( stepMeta, stepDataInterface, cnr, transMeta, trans );
}
 

@Override public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta, Trans trans ) {
  return new DummyTrans( stepMeta, stepDataInterface, copyNr, transMeta, trans );
}
 

@Override
public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr, TransMeta transMeta,
    Trans trans ) {
  return new SortRows( stepMeta, stepDataInterface, cnr, transMeta, trans );
}
 

@Override public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta, Trans trans ) {
  return new BeamConsume( stepMeta, stepDataInterface, copyNr, transMeta, trans );
}
 

public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr,
  TransMeta transMeta, Trans trans ) {
  return new CubeOutput( stepMeta, stepDataInterface, cnr, transMeta, trans );
}
 

@Override
public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr, TransMeta transMeta,
    Trans trans ) {
  return new TextFileInput( stepMeta, stepDataInterface, cnr, transMeta, trans );
}
 

public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr,
  TransMeta transMeta, Trans trans ) {
  return new RandomCCNumberGenerator( stepMeta, stepDataInterface, cnr, transMeta, trans );
}
 

public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr,
  TransMeta transMeta, Trans trans ) {
  return new AnalyticQuery( stepMeta, stepDataInterface, cnr, transMeta, trans );
}
 

public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr, TransMeta tr,
  Trans trans ) {
  return new Mapping( stepMeta, stepDataInterface, cnr, tr, trans );
}
 

@Override
public StepInterface getStep( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr,
  TransMeta transMeta, Trans trans ) {
  return new RulesAccumulator( stepMeta, stepDataInterface, copyNr, transMeta, trans );
}