@Test public void testExecuteWithNoAelSecurityInstalled() { ICapability aelSecurityCapability = mock( ICapability.class ); setCapability( aelSecurityCapability, SparkRunConfigurationExecutor.AEL_SECURITY_CAPABILITY_ID, false ); ICapability jaasCapability = mock( ICapability.class ); setCapability( jaasCapability, SparkRunConfigurationExecutor.JAAS_CAPABILITY_ID, false ); SparkRunConfiguration sparkRunConfiguration = new SparkRunConfiguration(); sparkRunConfiguration.setName( "Spark Configuration" ); TransExecutionConfiguration transExecutionConfiguration = new TransExecutionConfiguration(); sparkRunConfigurationExecutor .execute( sparkRunConfiguration, transExecutionConfiguration, abstractMeta, variableSpace ); verify( jaasCapability, never() ).isInstalled(); }
@Override public void execute( RunConfiguration runConfiguration, ExecutionConfiguration configuration, AbstractMeta meta, VariableSpace variableSpace ) { ICapability securityCapability = capabilityManager.getCapabilityById( AEL_SECURITY_CAPABILITY_ID ); ICapability jaasCapability = capabilityManager.getCapabilityById( JAAS_CAPABILITY_ID ); if ( securityCapability != null && securityCapability.isInstalled() ) { if ( jaasCapability != null && !jaasCapability.isInstalled() ) { jaasCapability.install(); } } ICapability capability = capabilityManager.getCapabilityById( ZOOKEEPER_CAPABILITY_ID ); if ( capability != null && !capability.isInstalled() ) { capability.install(); } SparkRunConfiguration sparkRunConfiguration = (SparkRunConfiguration) runConfiguration; String runConfigURL = Const.NVL( sparkRunConfiguration.getUrl(), "" ); URI uri = URI.create( runConfigURL.trim() ); String protocol = uri.getScheme(); String host = uri.getHost(); String port = uri.getPort() == -1 ? null : String.valueOf( uri.getPort() ); String version = variableSpace.getVariable( "KETTLE_AEL_PDI_DAEMON_VERSION", "2.0" ); boolean version2 = Const.toDouble( version, 1 ) >= 2; boolean serverMode = capabilityManager.getCapabilityById( PENTAHO_SERVER_CAPABILITY_ID ) != null; if ( version2 ) { variableSpace.setVariable( "engine.protocol", Const.NVL( protocol, DEFAULT_PROTOCOL ) ); variableSpace.setVariable( "engine.host", Const.NVL( host, DEFAULT_HOST ) ); variableSpace.setVariable( "engine.port", Const.NVL( port, DEFAULT_WEBSOCKET_PORT ) ); } if ( !serverMode ) { try { Configuration zookeeperConfiguration = configurationAdmin.getConfiguration( CONFIG_KEY ); Dictionary<String, Object> properties = zookeeperConfiguration.getProperties(); if ( properties != null ) { if ( !version2 ) { properties.put( "zookeeper.host", Const.NVL( host, DEFAULT_HOST ) ); properties.put( "zookeeper.port", Const.NVL( port, DEFAULT_ZOOKEEPER_PORT ) ); variableSpace.setVariable( "engine.protocol", null ); variableSpace.setVariable( "engine.host", null ); variableSpace.setVariable( "engine.port", null ); } else { properties.remove( "zookeeper.host" ); properties.remove( "zookeeper.port" ); } zookeeperConfiguration.update( properties ); } } catch ( IOException ioe ) { System.out.println( "Error occurred accessing configuration" ); } } variableSpace.setVariable( "engine", "remote" ); variableSpace.setVariable( "engine.remote", "spark" ); }
SparkRunConfigurationExecutor implements RunConfigurationExecutor { @Override public void execute( RunConfiguration runConfiguration, ExecutionConfiguration configuration, AbstractMeta meta, VariableSpace variableSpace ) { ICapability securityCapability = capabilityManager.getCapabilityById( AEL_SECURITY_CAPABILITY_ID ); ICapability jaasCapability = capabilityManager.getCapabilityById( JAAS_CAPABILITY_ID ); if ( securityCapability != null && securityCapability.isInstalled() ) { if ( jaasCapability != null && !jaasCapability.isInstalled() ) { jaasCapability.install(); } } ICapability capability = capabilityManager.getCapabilityById( ZOOKEEPER_CAPABILITY_ID ); if ( capability != null && !capability.isInstalled() ) { capability.install(); } SparkRunConfiguration sparkRunConfiguration = (SparkRunConfiguration) runConfiguration; String runConfigURL = Const.NVL( sparkRunConfiguration.getUrl(), "" ); URI uri = URI.create( runConfigURL.trim() ); String protocol = uri.getScheme(); String host = uri.getHost(); String port = uri.getPort() == -1 ? null : String.valueOf( uri.getPort() ); String version = variableSpace.getVariable( "KETTLE_AEL_PDI_DAEMON_VERSION", "2.0" ); boolean version2 = Const.toDouble( version, 1 ) >= 2; boolean serverMode = capabilityManager.getCapabilityById( PENTAHO_SERVER_CAPABILITY_ID ) != null; if ( version2 ) { variableSpace.setVariable( "engine.protocol", Const.NVL( protocol, DEFAULT_PROTOCOL ) ); variableSpace.setVariable( "engine.host", Const.NVL( host, DEFAULT_HOST ) ); variableSpace.setVariable( "engine.port", Const.NVL( port, DEFAULT_WEBSOCKET_PORT ) ); } if ( !serverMode ) { try { Configuration zookeeperConfiguration = configurationAdmin.getConfiguration( CONFIG_KEY ); Dictionary<String, Object> properties = zookeeperConfiguration.getProperties(); if ( properties != null ) { if ( !version2 ) { properties.put( "zookeeper.host", Const.NVL( host, DEFAULT_HOST ) ); properties.put( "zookeeper.port", Const.NVL( port, DEFAULT_ZOOKEEPER_PORT ) ); variableSpace.setVariable( "engine.protocol", null ); variableSpace.setVariable( "engine.host", null ); variableSpace.setVariable( "engine.port", null ); } else { properties.remove( "zookeeper.host" ); properties.remove( "zookeeper.port" ); } zookeeperConfiguration.update( properties ); } } catch ( IOException ioe ) { System.out.println( "Error occurred accessing configuration" ); } } variableSpace.setVariable( "engine", "remote" ); variableSpace.setVariable( "engine.remote", "spark" ); } }
SparkRunConfigurationExecutor implements RunConfigurationExecutor { @Override public void execute( RunConfiguration runConfiguration, ExecutionConfiguration configuration, AbstractMeta meta, VariableSpace variableSpace ) { ICapability securityCapability = capabilityManager.getCapabilityById( AEL_SECURITY_CAPABILITY_ID ); ICapability jaasCapability = capabilityManager.getCapabilityById( JAAS_CAPABILITY_ID ); if ( securityCapability != null && securityCapability.isInstalled() ) { if ( jaasCapability != null && !jaasCapability.isInstalled() ) { jaasCapability.install(); } } ICapability capability = capabilityManager.getCapabilityById( ZOOKEEPER_CAPABILITY_ID ); if ( capability != null && !capability.isInstalled() ) { capability.install(); } SparkRunConfiguration sparkRunConfiguration = (SparkRunConfiguration) runConfiguration; String runConfigURL = Const.NVL( sparkRunConfiguration.getUrl(), "" ); URI uri = URI.create( runConfigURL.trim() ); String protocol = uri.getScheme(); String host = uri.getHost(); String port = uri.getPort() == -1 ? null : String.valueOf( uri.getPort() ); String version = variableSpace.getVariable( "KETTLE_AEL_PDI_DAEMON_VERSION", "2.0" ); boolean version2 = Const.toDouble( version, 1 ) >= 2; boolean serverMode = capabilityManager.getCapabilityById( PENTAHO_SERVER_CAPABILITY_ID ) != null; if ( version2 ) { variableSpace.setVariable( "engine.protocol", Const.NVL( protocol, DEFAULT_PROTOCOL ) ); variableSpace.setVariable( "engine.host", Const.NVL( host, DEFAULT_HOST ) ); variableSpace.setVariable( "engine.port", Const.NVL( port, DEFAULT_WEBSOCKET_PORT ) ); } if ( !serverMode ) { try { Configuration zookeeperConfiguration = configurationAdmin.getConfiguration( CONFIG_KEY ); Dictionary<String, Object> properties = zookeeperConfiguration.getProperties(); if ( properties != null ) { if ( !version2 ) { properties.put( "zookeeper.host", Const.NVL( host, DEFAULT_HOST ) ); properties.put( "zookeeper.port", Const.NVL( port, DEFAULT_ZOOKEEPER_PORT ) ); variableSpace.setVariable( "engine.protocol", null ); variableSpace.setVariable( "engine.host", null ); variableSpace.setVariable( "engine.port", null ); } else { properties.remove( "zookeeper.host" ); properties.remove( "zookeeper.port" ); } zookeeperConfiguration.update( properties ); } } catch ( IOException ioe ) { System.out.println( "Error occurred accessing configuration" ); } } variableSpace.setVariable( "engine", "remote" ); variableSpace.setVariable( "engine.remote", "spark" ); } SparkRunConfigurationExecutor( ConfigurationAdmin configurationAdmin ); }
SparkRunConfigurationExecutor implements RunConfigurationExecutor { @Override public void execute( RunConfiguration runConfiguration, ExecutionConfiguration configuration, AbstractMeta meta, VariableSpace variableSpace ) { ICapability securityCapability = capabilityManager.getCapabilityById( AEL_SECURITY_CAPABILITY_ID ); ICapability jaasCapability = capabilityManager.getCapabilityById( JAAS_CAPABILITY_ID ); if ( securityCapability != null && securityCapability.isInstalled() ) { if ( jaasCapability != null && !jaasCapability.isInstalled() ) { jaasCapability.install(); } } ICapability capability = capabilityManager.getCapabilityById( ZOOKEEPER_CAPABILITY_ID ); if ( capability != null && !capability.isInstalled() ) { capability.install(); } SparkRunConfiguration sparkRunConfiguration = (SparkRunConfiguration) runConfiguration; String runConfigURL = Const.NVL( sparkRunConfiguration.getUrl(), "" ); URI uri = URI.create( runConfigURL.trim() ); String protocol = uri.getScheme(); String host = uri.getHost(); String port = uri.getPort() == -1 ? null : String.valueOf( uri.getPort() ); String version = variableSpace.getVariable( "KETTLE_AEL_PDI_DAEMON_VERSION", "2.0" ); boolean version2 = Const.toDouble( version, 1 ) >= 2; boolean serverMode = capabilityManager.getCapabilityById( PENTAHO_SERVER_CAPABILITY_ID ) != null; if ( version2 ) { variableSpace.setVariable( "engine.protocol", Const.NVL( protocol, DEFAULT_PROTOCOL ) ); variableSpace.setVariable( "engine.host", Const.NVL( host, DEFAULT_HOST ) ); variableSpace.setVariable( "engine.port", Const.NVL( port, DEFAULT_WEBSOCKET_PORT ) ); } if ( !serverMode ) { try { Configuration zookeeperConfiguration = configurationAdmin.getConfiguration( CONFIG_KEY ); Dictionary<String, Object> properties = zookeeperConfiguration.getProperties(); if ( properties != null ) { if ( !version2 ) { properties.put( "zookeeper.host", Const.NVL( host, DEFAULT_HOST ) ); properties.put( "zookeeper.port", Const.NVL( port, DEFAULT_ZOOKEEPER_PORT ) ); variableSpace.setVariable( "engine.protocol", null ); variableSpace.setVariable( "engine.host", null ); variableSpace.setVariable( "engine.port", null ); } else { properties.remove( "zookeeper.host" ); properties.remove( "zookeeper.port" ); } zookeeperConfiguration.update( properties ); } } catch ( IOException ioe ) { System.out.println( "Error occurred accessing configuration" ); } } variableSpace.setVariable( "engine", "remote" ); variableSpace.setVariable( "engine.remote", "spark" ); } SparkRunConfigurationExecutor( ConfigurationAdmin configurationAdmin ); @Override void execute( RunConfiguration runConfiguration, ExecutionConfiguration configuration,
                                 AbstractMeta meta, VariableSpace variableSpace ); }
SparkRunConfigurationExecutor implements RunConfigurationExecutor { @Override public void execute( RunConfiguration runConfiguration, ExecutionConfiguration configuration, AbstractMeta meta, VariableSpace variableSpace ) { ICapability securityCapability = capabilityManager.getCapabilityById( AEL_SECURITY_CAPABILITY_ID ); ICapability jaasCapability = capabilityManager.getCapabilityById( JAAS_CAPABILITY_ID ); if ( securityCapability != null && securityCapability.isInstalled() ) { if ( jaasCapability != null && !jaasCapability.isInstalled() ) { jaasCapability.install(); } } ICapability capability = capabilityManager.getCapabilityById( ZOOKEEPER_CAPABILITY_ID ); if ( capability != null && !capability.isInstalled() ) { capability.install(); } SparkRunConfiguration sparkRunConfiguration = (SparkRunConfiguration) runConfiguration; String runConfigURL = Const.NVL( sparkRunConfiguration.getUrl(), "" ); URI uri = URI.create( runConfigURL.trim() ); String protocol = uri.getScheme(); String host = uri.getHost(); String port = uri.getPort() == -1 ? null : String.valueOf( uri.getPort() ); String version = variableSpace.getVariable( "KETTLE_AEL_PDI_DAEMON_VERSION", "2.0" ); boolean version2 = Const.toDouble( version, 1 ) >= 2; boolean serverMode = capabilityManager.getCapabilityById( PENTAHO_SERVER_CAPABILITY_ID ) != null; if ( version2 ) { variableSpace.setVariable( "engine.protocol", Const.NVL( protocol, DEFAULT_PROTOCOL ) ); variableSpace.setVariable( "engine.host", Const.NVL( host, DEFAULT_HOST ) ); variableSpace.setVariable( "engine.port", Const.NVL( port, DEFAULT_WEBSOCKET_PORT ) ); } if ( !serverMode ) { try { Configuration zookeeperConfiguration = configurationAdmin.getConfiguration( CONFIG_KEY ); Dictionary<String, Object> properties = zookeeperConfiguration.getProperties(); if ( properties != null ) { if ( !version2 ) { properties.put( "zookeeper.host", Const.NVL( host, DEFAULT_HOST ) ); properties.put( "zookeeper.port", Const.NVL( port, DEFAULT_ZOOKEEPER_PORT ) ); variableSpace.setVariable( "engine.protocol", null ); variableSpace.setVariable( "engine.host", null ); variableSpace.setVariable( "engine.port", null ); } else { properties.remove( "zookeeper.host" ); properties.remove( "zookeeper.port" ); } zookeeperConfiguration.update( properties ); } } catch ( IOException ioe ) { System.out.println( "Error occurred accessing configuration" ); } } variableSpace.setVariable( "engine", "remote" ); variableSpace.setVariable( "engine.remote", "spark" ); } SparkRunConfigurationExecutor( ConfigurationAdmin configurationAdmin ); @Override void execute( RunConfiguration runConfiguration, ExecutionConfiguration configuration,
                                 AbstractMeta meta, VariableSpace variableSpace ); static String ZOOKEEPER_CAPABILITY_ID; static String PENTAHO_SERVER_CAPABILITY_ID; static String CONFIG_KEY; static String JAAS_CAPABILITY_ID; static String AEL_SECURITY_CAPABILITY_ID; static String DEFAULT_PROTOCOL; static String DEFAULT_HOST; static String DEFAULT_ZOOKEEPER_PORT; static String DEFAULT_WEBSOCKET_PORT; }