@Test(expected = FlumeException.class) public void testBadSource() throws Exception { properties.put("source.type", "exec"); EmbeddedAgentConfiguration.configure("test1", properties); }
@InterfaceAudience.Private @InterfaceStability.Unstable static Map<String, String> configure(String name, Map<String, String> properties) throws FlumeException { validate(name, properties); properties = new HashMap<String, String>(properties); if(!properties.containsKey(SOURCE_TYPE) || SOURCE_TYPE_EMBEDDED_ALIAS. equalsIgnoreCase(properties.get(SOURCE_TYPE))) { properties.put(SOURCE_TYPE, SOURCE_TYPE_EMBEDDED); } String sinkNames = properties.remove(SINKS); String sourceName = "source-" + name; String channelName = "channel-" + name; String sinkGroupName = "sink-group-" + name; Map<String, String> result = Maps.newHashMap(); Set<String> userProvidedKeys; result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES), sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS), sinkGroupName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, sourceName, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); userProvidedKeys = new HashSet<String>(properties.keySet()); for(String sink : sinkNames.split("\\s+")) { for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(sink + SEPERATOR)) { properties.remove(key); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, key), value); } } result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, sink, BasicConfigurationConstants.CONFIG_CHANNEL), channelName); } userProvidedKeys = new HashSet<String>(properties.keySet()); for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(SOURCE_PREFIX)) { key = key.replaceFirst(SOURCE, sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, key), value); } else if(key.startsWith(CHANNEL_PREFIX)) { key = key.replaceFirst(CHANNEL, channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS, key), value); } else if(key.startsWith(SINK_PROCESSOR_PREFIX)) { result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, key), value); } else { throw new FlumeException("Unknown configuration " + key); } } return result; }
EmbeddedAgentConfiguration { @InterfaceAudience.Private @InterfaceStability.Unstable static Map<String, String> configure(String name, Map<String, String> properties) throws FlumeException { validate(name, properties); properties = new HashMap<String, String>(properties); if(!properties.containsKey(SOURCE_TYPE) || SOURCE_TYPE_EMBEDDED_ALIAS. equalsIgnoreCase(properties.get(SOURCE_TYPE))) { properties.put(SOURCE_TYPE, SOURCE_TYPE_EMBEDDED); } String sinkNames = properties.remove(SINKS); String sourceName = "source-" + name; String channelName = "channel-" + name; String sinkGroupName = "sink-group-" + name; Map<String, String> result = Maps.newHashMap(); Set<String> userProvidedKeys; result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES), sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS), sinkGroupName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, sourceName, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); userProvidedKeys = new HashSet<String>(properties.keySet()); for(String sink : sinkNames.split("\\s+")) { for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(sink + SEPERATOR)) { properties.remove(key); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, key), value); } } result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, sink, BasicConfigurationConstants.CONFIG_CHANNEL), channelName); } userProvidedKeys = new HashSet<String>(properties.keySet()); for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(SOURCE_PREFIX)) { key = key.replaceFirst(SOURCE, sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, key), value); } else if(key.startsWith(CHANNEL_PREFIX)) { key = key.replaceFirst(CHANNEL, channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS, key), value); } else if(key.startsWith(SINK_PROCESSOR_PREFIX)) { result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, key), value); } else { throw new FlumeException("Unknown configuration " + key); } } return result; } }
EmbeddedAgentConfiguration { @InterfaceAudience.Private @InterfaceStability.Unstable static Map<String, String> configure(String name, Map<String, String> properties) throws FlumeException { validate(name, properties); properties = new HashMap<String, String>(properties); if(!properties.containsKey(SOURCE_TYPE) || SOURCE_TYPE_EMBEDDED_ALIAS. equalsIgnoreCase(properties.get(SOURCE_TYPE))) { properties.put(SOURCE_TYPE, SOURCE_TYPE_EMBEDDED); } String sinkNames = properties.remove(SINKS); String sourceName = "source-" + name; String channelName = "channel-" + name; String sinkGroupName = "sink-group-" + name; Map<String, String> result = Maps.newHashMap(); Set<String> userProvidedKeys; result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES), sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS), sinkGroupName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, sourceName, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); userProvidedKeys = new HashSet<String>(properties.keySet()); for(String sink : sinkNames.split("\\s+")) { for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(sink + SEPERATOR)) { properties.remove(key); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, key), value); } } result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, sink, BasicConfigurationConstants.CONFIG_CHANNEL), channelName); } userProvidedKeys = new HashSet<String>(properties.keySet()); for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(SOURCE_PREFIX)) { key = key.replaceFirst(SOURCE, sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, key), value); } else if(key.startsWith(CHANNEL_PREFIX)) { key = key.replaceFirst(CHANNEL, channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS, key), value); } else if(key.startsWith(SINK_PROCESSOR_PREFIX)) { result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, key), value); } else { throw new FlumeException("Unknown configuration " + key); } } return result; } private  EmbeddedAgentConfiguration(); }
EmbeddedAgentConfiguration { @InterfaceAudience.Private @InterfaceStability.Unstable static Map<String, String> configure(String name, Map<String, String> properties) throws FlumeException { validate(name, properties); properties = new HashMap<String, String>(properties); if(!properties.containsKey(SOURCE_TYPE) || SOURCE_TYPE_EMBEDDED_ALIAS. equalsIgnoreCase(properties.get(SOURCE_TYPE))) { properties.put(SOURCE_TYPE, SOURCE_TYPE_EMBEDDED); } String sinkNames = properties.remove(SINKS); String sourceName = "source-" + name; String channelName = "channel-" + name; String sinkGroupName = "sink-group-" + name; Map<String, String> result = Maps.newHashMap(); Set<String> userProvidedKeys; result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES), sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS), sinkGroupName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, sourceName, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); userProvidedKeys = new HashSet<String>(properties.keySet()); for(String sink : sinkNames.split("\\s+")) { for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(sink + SEPERATOR)) { properties.remove(key); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, key), value); } } result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, sink, BasicConfigurationConstants.CONFIG_CHANNEL), channelName); } userProvidedKeys = new HashSet<String>(properties.keySet()); for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(SOURCE_PREFIX)) { key = key.replaceFirst(SOURCE, sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, key), value); } else if(key.startsWith(CHANNEL_PREFIX)) { key = key.replaceFirst(CHANNEL, channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS, key), value); } else if(key.startsWith(SINK_PROCESSOR_PREFIX)) { result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, key), value); } else { throw new FlumeException("Unknown configuration " + key); } } return result; } private  EmbeddedAgentConfiguration();  }
EmbeddedAgentConfiguration { @InterfaceAudience.Private @InterfaceStability.Unstable static Map<String, String> configure(String name, Map<String, String> properties) throws FlumeException { validate(name, properties); properties = new HashMap<String, String>(properties); if(!properties.containsKey(SOURCE_TYPE) || SOURCE_TYPE_EMBEDDED_ALIAS. equalsIgnoreCase(properties.get(SOURCE_TYPE))) { properties.put(SOURCE_TYPE, SOURCE_TYPE_EMBEDDED); } String sinkNames = properties.remove(SINKS); String sourceName = "source-" + name; String channelName = "channel-" + name; String sinkGroupName = "sink-group-" + name; Map<String, String> result = Maps.newHashMap(); Set<String> userProvidedKeys; result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES), sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS), sinkGroupName); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, SINKS), sinkNames); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, sourceName, BasicConfigurationConstants.CONFIG_CHANNELS), channelName); userProvidedKeys = new HashSet<String>(properties.keySet()); for(String sink : sinkNames.split("\\s+")) { for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(sink + SEPERATOR)) { properties.remove(key); result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, key), value); } } result.put(join(name, BasicConfigurationConstants.CONFIG_SINKS, sink, BasicConfigurationConstants.CONFIG_CHANNEL), channelName); } userProvidedKeys = new HashSet<String>(properties.keySet()); for(String key : userProvidedKeys) { String value = properties.get(key); if(key.startsWith(SOURCE_PREFIX)) { key = key.replaceFirst(SOURCE, sourceName); result.put(join(name, BasicConfigurationConstants.CONFIG_SOURCES, key), value); } else if(key.startsWith(CHANNEL_PREFIX)) { key = key.replaceFirst(CHANNEL, channelName); result.put(join(name, BasicConfigurationConstants.CONFIG_CHANNELS, key), value); } else if(key.startsWith(SINK_PROCESSOR_PREFIX)) { result.put(join(name, BasicConfigurationConstants.CONFIG_SINKGROUPS, sinkGroupName, key), value); } else { throw new FlumeException("Unknown configuration " + key); } } return result; } private  EmbeddedAgentConfiguration();  static final String SEPERATOR; static final String SOURCE; static final String CHANNEL; static final String SINK_PROCESSOR; static final String SINKS; static final String SINKS_PREFIX; static final String SOURCE_TYPE; static final String SOURCE_PREFIX; static final String CHANNEL_TYPE; static final String CHANNEL_PREFIX; static final String SINK_PROCESSOR_TYPE; static final String SINK_PROCESSOR_PREFIX; static final String SOURCE_TYPE_EMBEDDED; static final String CHANNEL_TYPE_MEMORY; static final String CHANNEL_TYPE_FILE; static final String SINK_TYPE_AVRO; static final String SINK_PROCESSOR_TYPE_DEFAULT; static final String SINK_PROCESSOR_TYPE_FAILOVER; static final String SINK_PROCESSOR_TYPE_LOAD_BALANCE; }