@Test(expected = ConfigurationException.class) public void testFLUME1847() throws Exception { Context context = new Context(); context.put("type", "something"); SourceConfiguration sourceConfig = new SourceConfiguration("src"); sourceConfig.configure(context); }
public void configure(Context context) throws ConfigurationException { super.configure(context); try { String channelList = context.getString( BasicConfigurationConstants.CONFIG_CHANNELS); if (channelList != null) { this.channels = new HashSet<String>(Arrays.asList(channelList.split("\\s+"))); } if (channels.isEmpty()) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNEL.getComponentType(), FlumeConfigurationErrorType.PROPERTY_VALUE_NULL, ErrorOrWarning.ERROR)); throw new ConfigurationException("No channels set for " + this.getComponentName()); } Map<String, String> selectorParams = context.getSubProperties( BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX); String selType; if (selectorParams != null && !selectorParams.isEmpty()) { selType = selectorParams.get(BasicConfigurationConstants.CONFIG_TYPE); } else { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } if (selType == null || selType.isEmpty()) { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } ChannelSelectorType selectorType = this.getKnownChannelSelector(selType); Context selectorContext = new Context(); selectorContext.putAll(selectorParams); String config = null; if (selectorType == null) { config = selectorContext.getString( BasicConfigurationConstants.CONFIG_CONFIG); if (config == null || config.isEmpty()) { config = "OTHER"; } } else { config = selectorType.toString().toUpperCase(); } this.selectorConf = (ChannelSelectorConfiguration) ComponentConfigurationFactory .create(ComponentType.CHANNELSELECTOR.getComponentType(), config, ComponentType.CHANNELSELECTOR); selectorConf.setChannels(channels); selectorConf.configure(selectorContext); } catch (Exception e) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNELSELECTOR.getComponentType(), FlumeConfigurationErrorType.CONFIG_ERROR, ErrorOrWarning.ERROR)); throw new ConfigurationException("Failed to configure component!", e); } }
SourceConfiguration extends ComponentConfiguration { public void configure(Context context) throws ConfigurationException { super.configure(context); try { String channelList = context.getString( BasicConfigurationConstants.CONFIG_CHANNELS); if (channelList != null) { this.channels = new HashSet<String>(Arrays.asList(channelList.split("\\s+"))); } if (channels.isEmpty()) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNEL.getComponentType(), FlumeConfigurationErrorType.PROPERTY_VALUE_NULL, ErrorOrWarning.ERROR)); throw new ConfigurationException("No channels set for " + this.getComponentName()); } Map<String, String> selectorParams = context.getSubProperties( BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX); String selType; if (selectorParams != null && !selectorParams.isEmpty()) { selType = selectorParams.get(BasicConfigurationConstants.CONFIG_TYPE); } else { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } if (selType == null || selType.isEmpty()) { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } ChannelSelectorType selectorType = this.getKnownChannelSelector(selType); Context selectorContext = new Context(); selectorContext.putAll(selectorParams); String config = null; if (selectorType == null) { config = selectorContext.getString( BasicConfigurationConstants.CONFIG_CONFIG); if (config == null || config.isEmpty()) { config = "OTHER"; } } else { config = selectorType.toString().toUpperCase(); } this.selectorConf = (ChannelSelectorConfiguration) ComponentConfigurationFactory .create(ComponentType.CHANNELSELECTOR.getComponentType(), config, ComponentType.CHANNELSELECTOR); selectorConf.setChannels(channels); selectorConf.configure(selectorContext); } catch (Exception e) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNELSELECTOR.getComponentType(), FlumeConfigurationErrorType.CONFIG_ERROR, ErrorOrWarning.ERROR)); throw new ConfigurationException("Failed to configure component!", e); } } }
SourceConfiguration extends ComponentConfiguration { public void configure(Context context) throws ConfigurationException { super.configure(context); try { String channelList = context.getString( BasicConfigurationConstants.CONFIG_CHANNELS); if (channelList != null) { this.channels = new HashSet<String>(Arrays.asList(channelList.split("\\s+"))); } if (channels.isEmpty()) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNEL.getComponentType(), FlumeConfigurationErrorType.PROPERTY_VALUE_NULL, ErrorOrWarning.ERROR)); throw new ConfigurationException("No channels set for " + this.getComponentName()); } Map<String, String> selectorParams = context.getSubProperties( BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX); String selType; if (selectorParams != null && !selectorParams.isEmpty()) { selType = selectorParams.get(BasicConfigurationConstants.CONFIG_TYPE); } else { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } if (selType == null || selType.isEmpty()) { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } ChannelSelectorType selectorType = this.getKnownChannelSelector(selType); Context selectorContext = new Context(); selectorContext.putAll(selectorParams); String config = null; if (selectorType == null) { config = selectorContext.getString( BasicConfigurationConstants.CONFIG_CONFIG); if (config == null || config.isEmpty()) { config = "OTHER"; } } else { config = selectorType.toString().toUpperCase(); } this.selectorConf = (ChannelSelectorConfiguration) ComponentConfigurationFactory .create(ComponentType.CHANNELSELECTOR.getComponentType(), config, ComponentType.CHANNELSELECTOR); selectorConf.setChannels(channels); selectorConf.configure(selectorContext); } catch (Exception e) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNELSELECTOR.getComponentType(), FlumeConfigurationErrorType.CONFIG_ERROR, ErrorOrWarning.ERROR)); throw new ConfigurationException("Failed to configure component!", e); } } SourceConfiguration(String componentName); }
SourceConfiguration extends ComponentConfiguration { public void configure(Context context) throws ConfigurationException { super.configure(context); try { String channelList = context.getString( BasicConfigurationConstants.CONFIG_CHANNELS); if (channelList != null) { this.channels = new HashSet<String>(Arrays.asList(channelList.split("\\s+"))); } if (channels.isEmpty()) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNEL.getComponentType(), FlumeConfigurationErrorType.PROPERTY_VALUE_NULL, ErrorOrWarning.ERROR)); throw new ConfigurationException("No channels set for " + this.getComponentName()); } Map<String, String> selectorParams = context.getSubProperties( BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX); String selType; if (selectorParams != null && !selectorParams.isEmpty()) { selType = selectorParams.get(BasicConfigurationConstants.CONFIG_TYPE); } else { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } if (selType == null || selType.isEmpty()) { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } ChannelSelectorType selectorType = this.getKnownChannelSelector(selType); Context selectorContext = new Context(); selectorContext.putAll(selectorParams); String config = null; if (selectorType == null) { config = selectorContext.getString( BasicConfigurationConstants.CONFIG_CONFIG); if (config == null || config.isEmpty()) { config = "OTHER"; } } else { config = selectorType.toString().toUpperCase(); } this.selectorConf = (ChannelSelectorConfiguration) ComponentConfigurationFactory .create(ComponentType.CHANNELSELECTOR.getComponentType(), config, ComponentType.CHANNELSELECTOR); selectorConf.setChannels(channels); selectorConf.configure(selectorContext); } catch (Exception e) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNELSELECTOR.getComponentType(), FlumeConfigurationErrorType.CONFIG_ERROR, ErrorOrWarning.ERROR)); throw new ConfigurationException("Failed to configure component!", e); } } SourceConfiguration(String componentName); Set<String> getChannels(); ChannelSelectorConfiguration getSelectorConfiguration(); void configure(Context context); @Override String toString(int indentCount); }
SourceConfiguration extends ComponentConfiguration { public void configure(Context context) throws ConfigurationException { super.configure(context); try { String channelList = context.getString( BasicConfigurationConstants.CONFIG_CHANNELS); if (channelList != null) { this.channels = new HashSet<String>(Arrays.asList(channelList.split("\\s+"))); } if (channels.isEmpty()) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNEL.getComponentType(), FlumeConfigurationErrorType.PROPERTY_VALUE_NULL, ErrorOrWarning.ERROR)); throw new ConfigurationException("No channels set for " + this.getComponentName()); } Map<String, String> selectorParams = context.getSubProperties( BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX); String selType; if (selectorParams != null && !selectorParams.isEmpty()) { selType = selectorParams.get(BasicConfigurationConstants.CONFIG_TYPE); } else { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } if (selType == null || selType.isEmpty()) { selType = ChannelSelectorConfigurationType.REPLICATING.toString(); } ChannelSelectorType selectorType = this.getKnownChannelSelector(selType); Context selectorContext = new Context(); selectorContext.putAll(selectorParams); String config = null; if (selectorType == null) { config = selectorContext.getString( BasicConfigurationConstants.CONFIG_CONFIG); if (config == null || config.isEmpty()) { config = "OTHER"; } } else { config = selectorType.toString().toUpperCase(); } this.selectorConf = (ChannelSelectorConfiguration) ComponentConfigurationFactory .create(ComponentType.CHANNELSELECTOR.getComponentType(), config, ComponentType.CHANNELSELECTOR); selectorConf.setChannels(channels); selectorConf.configure(selectorContext); } catch (Exception e) { errors.add(new FlumeConfigurationError(componentName, ComponentType.CHANNELSELECTOR.getComponentType(), FlumeConfigurationErrorType.CONFIG_ERROR, ErrorOrWarning.ERROR)); throw new ConfigurationException("Failed to configure component!", e); } } SourceConfiguration(String componentName); Set<String> getChannels(); ChannelSelectorConfiguration getSelectorConfiguration(); void configure(Context context); @Override String toString(int indentCount);  }