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 strippedName = name.replaceAll("\\s+",""); String sourceName = "source-" + strippedName; String channelName = "channel-" + strippedName; String sinkGroupName = "sink-group-" + strippedName; Map<String, String> result = Maps.newHashMap(); 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); Set<String> 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_SPILLABLEMEMORY; 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; }
@Test public void testFullSourceType() throws Exception { doTestExcepted(EmbeddedAgentConfiguration.configure("test1", properties)); }
@Test public void testMissingSourceType() throws Exception { Assert.assertNotNull(properties.remove("source.type")); doTestExcepted(EmbeddedAgentConfiguration.configure("test1", properties)); }
@Test public void testShortSourceType() throws Exception { properties.put("source.type", "EMBEDDED"); doTestExcepted(EmbeddedAgentConfiguration.configure("test1", properties)); }
@Test(expected = FlumeException.class) public void testBadSource() throws Exception { properties.put("source.type", "exec"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testBadChannel() throws Exception { properties.put("channel.type", "jdbc"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testBadSink() throws Exception { properties.put("sink1.type", "hbase"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testBadSinkProcessor() throws Exception { properties.put("processor.type", "bad"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testNoChannel() throws Exception { properties.remove("channel.type"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testNoSink() throws Exception { properties.remove("sink2.type"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testNoSinkProcessor() throws Exception { properties.remove("processor.type"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testBadKey() throws Exception { properties.put("bad.key.name", "bad"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testSinkNamedLikeSource() throws Exception { properties.put("sinks", "source"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testSinkNamedLikeChannel() throws Exception { properties.put("sinks", "channel"); EmbeddedAgentConfiguration.configure("test1", properties); }
@Test(expected = FlumeException.class) public void testSinkNamedLikeProcessor() throws Exception { properties.put("sinks", "processor"); EmbeddedAgentConfiguration.configure("test1", properties); }