@Override
  public boolean init( StepMetaInterface stepMetaInterface, StepDataInterface stepDataInterface ) {
    Preconditions.checkNotNull( stepMetaInterface );
    variablizedStepMeta = (BaseStreamStepMeta) stepMetaInterface;
    variablizedStepMeta.setParentStepMeta( getStepMeta() );
    variablizedStepMeta.setFileName( variablizedStepMeta.getTransformationPath() );

    boolean superInit = super.init( stepMetaInterface, stepDataInterface );

    try {
      TransMeta transMeta = TransExecutorMeta
        .loadMappingMeta( variablizedStepMeta, getTransMeta().getRepository(), getTransMeta().getMetaStore(),
          getParentVariableSpace() );
      variablizedStepMeta = (BaseStreamStepMeta) variablizedStepMeta.withVariables( this );
      subtransExecutor = new SubtransExecutor( getStepname(),
        getTrans(), transMeta, true,
        new TransExecutorParameters(), variablizedStepMeta.getSubStep(), getPrefetchCount() );

    } catch ( KettleException e ) {
      log.logError( e.getLocalizedMessage(), e );
      return false;
    }

    List<CheckResultInterface> remarks = new ArrayList<>();
    variablizedStepMeta.check(
      remarks, getTransMeta(), variablizedStepMeta.getParentStepMeta(),
      null, null, null, null, //these parameters are not used inside the method
      variables, getRepository(), getMetaStore() );
    boolean errorsPresent =
      remarks.stream().filter( result -> result.getType() == CheckResultInterface.TYPE_RESULT_ERROR )
        .peek( result -> logError( result.getText() ) )
        .count() > 0;
    if ( errorsPresent ) {
      return false;
    }
    return superInit;
  }
@Test
  public void testInitFilenameSubstitution() throws IOException {
    // verifies that filename resolution uses the parents ${Internal.Entry.Current.Directory}.
    // Necessary since the Current.Directory may change when running non-locally.
    // Variables should all be set in variableizedStepMeta after init, with the caveat that
    // the substrans location must be set using the parents Current.Directory.
    File testFile = File.createTempFile( "testInitFilenameSubstitution", ".ktr",
      folder.getRoot() );
    try ( PrintWriter pw = new PrintWriter( testFile ) ) {
      // empty subtrans definition
      pw.write( "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
        + "<transformation/>" );
    }

    when( meta.getParentStepMeta() ).thenReturn( parentStepMeta );
    when( meta.withVariables( baseStreamStep ) ).thenReturn( metaWithVariables );
    baseStreamStep.getParentVariableSpace()
      .setVariable( "Internal.Entry.Current.Directory",
        testFile.getParentFile().getAbsolutePath() );

    when( meta.getSpecificationMethod() ).thenReturn( ObjectLocationSpecificationMethod.FILENAME );
    when( meta.getFileName() ).thenReturn( "${Internal.Entry.Current.Directory}/" + testFile.getName() );

    assertTrue( baseStreamStep.init( meta, stepData ) );
    assertThat( baseStreamStep.variablizedStepMeta, equalTo( metaWithVariables ) );
  }