@Test public void testReplaceHadoopClusterToFileSystemURL() throws KettleException, MetaStoreException { TransMeta origTransMeta = new TransMeta(); Variables variables = new Variables(); String inputUrl = "hc: String outputUrl = "hc: TransMeta transMeta = spy( origTransMeta ); IMetaStore metaStore = mock( IMetaStore.class ); NamedClusterServiceOsgi namedClusterServiceOsgi = mock( NamedClusterServiceOsgi.class ); NamedClusterOsgi namedClusterOsgi = mock( NamedClusterOsgi.class ); NamedClusterEmbedManager namedClusterEmbedManager = mock( NamedClusterEmbedManager.class ); transMeta.setParentVariableSpace( variables ); transMeta.setMetaStore( metaStore ); transMeta.setNamedClusterServiceOsgi( namedClusterServiceOsgi ); doReturn( transMeta ).when( transMeta ).realClone( false ); doReturn( namedClusterEmbedManager ).when( transMeta ).getNamedClusterEmbedManager( ); doReturn( namedClusterOsgi ).when( namedClusterServiceOsgi ).read( "CDH511Unsecure", metaStore ); doReturn( null ).when( namedClusterServiceOsgi ).read( "CDH512Unsecure", metaStore ); when( namedClusterOsgi.processURLsubstitution( inputUrl, metaStore, variables ) ).thenReturn( "hdfs: when( namedClusterOsgi.processURLsubstitution( outputUrl, metaStore, variables ) ).thenReturn( "hdfs: TextFileInputMeta textFileInputMeta = new TextFileInputMeta(); String[] inputFiles = new String[6]; inputFiles[0] = "hc: inputFiles[1] = "hc: inputFiles[2] = "hdfs: inputFiles[3] = "hdfs: inputFiles[4] = "C:/Users/testuser/Downloads/testfile"; inputFiles[5] = "file: textFileInputMeta.setFileNameForTest( inputFiles ); TextFileOutputMeta textFileOutputMeta = new TextFileOutputMeta(); textFileOutputMeta.setFileName( "hc: textFileOutputMeta.allocate( 0 ); StepMeta textFileInput = new StepMeta( "TextFileInput", textFileInputMeta ); StepMeta textFileOutput = new StepMeta( "TextFileOutput", textFileOutputMeta ); transMeta.addStep( textFileInput ); transMeta.addStep( textFileOutput ); transMeta.addTransHop( new TransHopMeta( textFileInput, textFileOutput ) ); TransMeta cloneMeta = getTransMetaFromTrans( TransMetaConverter.convert( transMeta ) ); for ( StepMeta stepMeta : cloneMeta.getSteps() ) { if ( stepMeta.getStepMetaInterface() instanceof TextFileInputMeta ) { TextFileInputMeta meta = (TextFileInputMeta) stepMeta.getStepMetaInterface(); String[] files = meta.getFileName(); for ( int i = 0; i < files.length; i++ ) { if ( files[i] != null ) { assertFalse( files[i].contains( "hc: } } } else if ( stepMeta.getStepMetaInterface() instanceof TextFileOutputMeta ) { TextFileOutputMeta meta = (TextFileOutputMeta) stepMeta.getStepMetaInterface(); String filename = meta.getFileName(); if ( filename != null ) { assertFalse( filename.contains( "hc: } } } }
public static Transformation convert( TransMeta transMeta ) { final org.pentaho.di.engine.model.Transformation transformation = new org.pentaho.di.engine.model.Transformation( createTransformationId( transMeta ) ); try { TransMeta copyTransMeta = (TransMeta) transMeta.realClone( false ); cleanupDisabledHops( copyTransMeta ); disableLazyConversion( copyTransMeta ); searchAndReplaceHadoopClusterWithFileSystemURL( copyTransMeta ); copyTransMeta.getSteps().forEach( createOperation( transformation ) ); findHops( copyTransMeta, hop -> true ).forEach( createHop( transformation ) ); transformation.setConfig( TRANS_META_CONF_KEY, copyTransMeta.getXML() ); transformation.setConfig( TRANS_META_NAME_CONF_KEY, Optional.ofNullable( transMeta.getName() ).orElse( TRANS_DEFAULT_NAME ) ); Map<String, Transformation> subTransformations = copyTransMeta.getResourceDependencies().stream() .flatMap( resourceReference -> resourceReference.getEntries().stream() ) .filter( entry -> ResourceEntry.ResourceType.ACTIONFILE.equals( entry.getResourcetype() ) ) .collect( toMap( ResourceEntry::getResource, entry -> { try { return convert( new TransMeta( entry.getResource(), copyTransMeta.getParentVariableSpace() ) ); } catch ( KettleXMLException | KettleMissingPluginsException e ) { throw new RuntimeException( e ); } } ) ); transformation.setConfig( SUB_TRANSFORMATIONS_KEY, (Serializable) subTransformations ); } catch ( KettleException e ) { Throwables.propagate( e ); } return transformation; }
TransMetaConverter { public static Transformation convert( TransMeta transMeta ) { final org.pentaho.di.engine.model.Transformation transformation = new org.pentaho.di.engine.model.Transformation( createTransformationId( transMeta ) ); try { TransMeta copyTransMeta = (TransMeta) transMeta.realClone( false ); cleanupDisabledHops( copyTransMeta ); disableLazyConversion( copyTransMeta ); searchAndReplaceHadoopClusterWithFileSystemURL( copyTransMeta ); copyTransMeta.getSteps().forEach( createOperation( transformation ) ); findHops( copyTransMeta, hop -> true ).forEach( createHop( transformation ) ); transformation.setConfig( TRANS_META_CONF_KEY, copyTransMeta.getXML() ); transformation.setConfig( TRANS_META_NAME_CONF_KEY, Optional.ofNullable( transMeta.getName() ).orElse( TRANS_DEFAULT_NAME ) ); Map<String, Transformation> subTransformations = copyTransMeta.getResourceDependencies().stream() .flatMap( resourceReference -> resourceReference.getEntries().stream() ) .filter( entry -> ResourceEntry.ResourceType.ACTIONFILE.equals( entry.getResourcetype() ) ) .collect( toMap( ResourceEntry::getResource, entry -> { try { return convert( new TransMeta( entry.getResource(), copyTransMeta.getParentVariableSpace() ) ); } catch ( KettleXMLException | KettleMissingPluginsException e ) { throw new RuntimeException( e ); } } ) ); transformation.setConfig( SUB_TRANSFORMATIONS_KEY, (Serializable) subTransformations ); } catch ( KettleException e ) { Throwables.propagate( e ); } return transformation; } }
TransMetaConverter { public static Transformation convert( TransMeta transMeta ) { final org.pentaho.di.engine.model.Transformation transformation = new org.pentaho.di.engine.model.Transformation( createTransformationId( transMeta ) ); try { TransMeta copyTransMeta = (TransMeta) transMeta.realClone( false ); cleanupDisabledHops( copyTransMeta ); disableLazyConversion( copyTransMeta ); searchAndReplaceHadoopClusterWithFileSystemURL( copyTransMeta ); copyTransMeta.getSteps().forEach( createOperation( transformation ) ); findHops( copyTransMeta, hop -> true ).forEach( createHop( transformation ) ); transformation.setConfig( TRANS_META_CONF_KEY, copyTransMeta.getXML() ); transformation.setConfig( TRANS_META_NAME_CONF_KEY, Optional.ofNullable( transMeta.getName() ).orElse( TRANS_DEFAULT_NAME ) ); Map<String, Transformation> subTransformations = copyTransMeta.getResourceDependencies().stream() .flatMap( resourceReference -> resourceReference.getEntries().stream() ) .filter( entry -> ResourceEntry.ResourceType.ACTIONFILE.equals( entry.getResourcetype() ) ) .collect( toMap( ResourceEntry::getResource, entry -> { try { return convert( new TransMeta( entry.getResource(), copyTransMeta.getParentVariableSpace() ) ); } catch ( KettleXMLException | KettleMissingPluginsException e ) { throw new RuntimeException( e ); } } ) ); transformation.setConfig( SUB_TRANSFORMATIONS_KEY, (Serializable) subTransformations ); } catch ( KettleException e ) { Throwables.propagate( e ); } return transformation; }  }
TransMetaConverter { public static Transformation convert( TransMeta transMeta ) { final org.pentaho.di.engine.model.Transformation transformation = new org.pentaho.di.engine.model.Transformation( createTransformationId( transMeta ) ); try { TransMeta copyTransMeta = (TransMeta) transMeta.realClone( false ); cleanupDisabledHops( copyTransMeta ); disableLazyConversion( copyTransMeta ); searchAndReplaceHadoopClusterWithFileSystemURL( copyTransMeta ); copyTransMeta.getSteps().forEach( createOperation( transformation ) ); findHops( copyTransMeta, hop -> true ).forEach( createHop( transformation ) ); transformation.setConfig( TRANS_META_CONF_KEY, copyTransMeta.getXML() ); transformation.setConfig( TRANS_META_NAME_CONF_KEY, Optional.ofNullable( transMeta.getName() ).orElse( TRANS_DEFAULT_NAME ) ); Map<String, Transformation> subTransformations = copyTransMeta.getResourceDependencies().stream() .flatMap( resourceReference -> resourceReference.getEntries().stream() ) .filter( entry -> ResourceEntry.ResourceType.ACTIONFILE.equals( entry.getResourcetype() ) ) .collect( toMap( ResourceEntry::getResource, entry -> { try { return convert( new TransMeta( entry.getResource(), copyTransMeta.getParentVariableSpace() ) ); } catch ( KettleXMLException | KettleMissingPluginsException e ) { throw new RuntimeException( e ); } } ) ); transformation.setConfig( SUB_TRANSFORMATIONS_KEY, (Serializable) subTransformations ); } catch ( KettleException e ) { Throwables.propagate( e ); } return transformation; }  static Transformation convert( TransMeta transMeta ); }
TransMetaConverter { public static Transformation convert( TransMeta transMeta ) { final org.pentaho.di.engine.model.Transformation transformation = new org.pentaho.di.engine.model.Transformation( createTransformationId( transMeta ) ); try { TransMeta copyTransMeta = (TransMeta) transMeta.realClone( false ); cleanupDisabledHops( copyTransMeta ); disableLazyConversion( copyTransMeta ); searchAndReplaceHadoopClusterWithFileSystemURL( copyTransMeta ); copyTransMeta.getSteps().forEach( createOperation( transformation ) ); findHops( copyTransMeta, hop -> true ).forEach( createHop( transformation ) ); transformation.setConfig( TRANS_META_CONF_KEY, copyTransMeta.getXML() ); transformation.setConfig( TRANS_META_NAME_CONF_KEY, Optional.ofNullable( transMeta.getName() ).orElse( TRANS_DEFAULT_NAME ) ); Map<String, Transformation> subTransformations = copyTransMeta.getResourceDependencies().stream() .flatMap( resourceReference -> resourceReference.getEntries().stream() ) .filter( entry -> ResourceEntry.ResourceType.ACTIONFILE.equals( entry.getResourcetype() ) ) .collect( toMap( ResourceEntry::getResource, entry -> { try { return convert( new TransMeta( entry.getResource(), copyTransMeta.getParentVariableSpace() ) ); } catch ( KettleXMLException | KettleMissingPluginsException e ) { throw new RuntimeException( e ); } } ) ); transformation.setConfig( SUB_TRANSFORMATIONS_KEY, (Serializable) subTransformations ); } catch ( KettleException e ) { Throwables.propagate( e ); } return transformation; }  static Transformation convert( TransMeta transMeta ); static final String TRANS_META_CONF_KEY; static final String TRANS_META_NAME_CONF_KEY; static final String SUB_TRANSFORMATIONS_KEY; static final String STEP_META_CONF_KEY; static final String TRANS_DEFAULT_NAME; }