AbstractConfigurationProvider implements ConfigurationProvider { public MaterializedConfiguration getConfiguration() { MaterializedConfiguration conf = new SimpleMaterializedConfiguration(); FlumeConfiguration fconfig = getFlumeConfiguration(); AgentConfiguration agentConf = fconfig.getConfigurationFor(getAgentName()); if (agentConf != null) { Map<String, ChannelComponent> channelComponentMap = Maps.newHashMap(); Map<String, SourceRunner> sourceRunnerMap = Maps.newHashMap(); Map<String, SinkRunner> sinkRunnerMap = Maps.newHashMap(); try { loadChannels(agentConf, channelComponentMap); loadSources(agentConf, channelComponentMap, sourceRunnerMap); loadSinks(agentConf, channelComponentMap, sinkRunnerMap); Set<String> channelNames = new HashSet<String>(channelComponentMap.keySet()); for (String channelName : channelNames) { ChannelComponent channelComponent = channelComponentMap.get(channelName); if (channelComponent.components.isEmpty()) { LOGGER.warn(String.format("Channel %s has no components connected" + " and has been removed.", channelName)); channelComponentMap.remove(channelName); Map<String, Channel> nameChannelMap = channelCache.get(channelComponent.channel.getClass()); if (nameChannelMap != null) { nameChannelMap.remove(channelName); } } else { LOGGER.info(String.format("Channel %s connected to %s", channelName, channelComponent.components.toString())); conf.addChannel(channelName, channelComponent.channel); } } for (Map.Entry<String, SourceRunner> entry : sourceRunnerMap.entrySet()) { conf.addSourceRunner(entry.getKey(), entry.getValue()); } for (Map.Entry<String, SinkRunner> entry : sinkRunnerMap.entrySet()) { conf.addSinkRunner(entry.getKey(), entry.getValue()); } } catch (InstantiationException ex) { LOGGER.error("Failed to instantiate component", ex); } finally { channelComponentMap.clear(); sourceRunnerMap.clear(); sinkRunnerMap.clear(); } } else { LOGGER.warn("No configuration found for this host:{}", getAgentName()); } return conf; } AbstractConfigurationProvider(String agentName); MaterializedConfiguration getConfiguration(); String getAgentName();  }
@Test public void testDispoableChannel() throws Exception { String agentName = "agent1"; Map<String, String> properties = getPropertiesForChannel(agentName, DisposableChannel.class.getName()); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, properties); MaterializedConfiguration config1 = provider.getConfiguration(); Channel channel1 = config1.getChannels().values().iterator().next(); Assert.assertTrue(channel1 instanceof DisposableChannel); MaterializedConfiguration config2 = provider.getConfiguration(); Channel channel2 = config2.getChannels().values().iterator().next(); Assert.assertTrue(channel2 instanceof DisposableChannel); Assert.assertNotSame(channel1, channel2); }
@Test public void testReusableChannel() throws Exception { String agentName = "agent1"; Map<String, String> properties = getPropertiesForChannel(agentName, RecyclableChannel.class.getName()); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, properties); MaterializedConfiguration config1 = provider.getConfiguration(); Channel channel1 = config1.getChannels().values().iterator().next(); Assert.assertTrue(channel1 instanceof RecyclableChannel); MaterializedConfiguration config2 = provider.getConfiguration(); Channel channel2 = config2.getChannels().values().iterator().next(); Assert.assertTrue(channel2 instanceof RecyclableChannel); Assert.assertSame(channel1, channel2); }
@Test public void testUnspecifiedChannel() throws Exception { String agentName = "agent1"; Map<String, String> properties = getPropertiesForChannel(agentName, UnspecifiedChannel.class.getName()); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, properties); MaterializedConfiguration config1 = provider.getConfiguration(); Channel channel1 = config1.getChannels().values().iterator().next(); Assert.assertTrue(channel1 instanceof UnspecifiedChannel); MaterializedConfiguration config2 = provider.getConfiguration(); Channel channel2 = config2.getChannels().values().iterator().next(); Assert.assertTrue(channel2 instanceof UnspecifiedChannel); Assert.assertSame(channel1, channel2); }
@Test public void testReusableChannelNotReusedLater() throws Exception { String agentName = "agent1"; Map<String, String> propertiesReusable = getPropertiesForChannel(agentName, RecyclableChannel.class .getName()); Map<String, String> propertiesDispoable = getPropertiesForChannel(agentName, DisposableChannel.class .getName()); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, propertiesReusable); MaterializedConfiguration config1 = provider.getConfiguration(); Channel channel1 = config1.getChannels().values().iterator().next(); Assert.assertTrue(channel1 instanceof RecyclableChannel); provider.setProperties(propertiesDispoable); MaterializedConfiguration config2 = provider.getConfiguration(); Channel channel2 = config2.getChannels().values().iterator().next(); Assert.assertTrue(channel2 instanceof DisposableChannel); provider.setProperties(propertiesReusable); MaterializedConfiguration config3 = provider.getConfiguration(); Channel channel3 = config3.getChannels().values().iterator().next(); Assert.assertTrue(channel3 instanceof RecyclableChannel); Assert.assertNotSame(channel1, channel3); }
@Test public void testSourceThrowsExceptionDuringConfiguration() throws Exception { String agentName = "agent1"; String sourceType = UnconfigurableSource.class.getName(); String channelType = "memory"; String sinkType = "null"; Map<String, String> properties = getProperties(agentName, sourceType, channelType, sinkType); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, properties); MaterializedConfiguration config = provider.getConfiguration(); Assert.assertTrue(config.getSourceRunners().size() == 0); Assert.assertTrue(config.getChannels().size() == 1); Assert.assertTrue(config.getSinkRunners().size() == 1); }
@Test public void testChannelThrowsExceptionDuringConfiguration() throws Exception { String agentName = "agent1"; String sourceType = "seq"; String channelType = UnconfigurableChannel.class.getName(); String sinkType = "null"; Map<String, String> properties = getProperties(agentName, sourceType, channelType, sinkType); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, properties); MaterializedConfiguration config = provider.getConfiguration(); Assert.assertTrue(config.getSourceRunners().size() == 0); Assert.assertTrue(config.getChannels().size() == 0); Assert.assertTrue(config.getSinkRunners().size() == 0); }
@Test public void testSinkThrowsExceptionDuringConfiguration() throws Exception { String agentName = "agent1"; String sourceType = "seq"; String channelType = "memory"; String sinkType = UnconfigurableSink.class.getName(); Map<String, String> properties = getProperties(agentName, sourceType, channelType, sinkType); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, properties); MaterializedConfiguration config = provider.getConfiguration(); Assert.assertTrue(config.getSourceRunners().size() == 1); Assert.assertTrue(config.getChannels().size() == 1); Assert.assertTrue(config.getSinkRunners().size() == 0); }
@Test public void testSourceAndSinkThrowExceptionDuringConfiguration() throws Exception { String agentName = "agent1"; String sourceType = UnconfigurableSource.class.getName(); String channelType = "memory"; String sinkType = UnconfigurableSink.class.getName(); Map<String, String> properties = getProperties(agentName, sourceType, channelType, sinkType); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, properties); MaterializedConfiguration config = provider.getConfiguration(); Assert.assertTrue(config.getSourceRunners().size() == 0); Assert.assertTrue(config.getChannels().size() == 0); Assert.assertTrue(config.getSinkRunners().size() == 0); }
@Test public void testSinkSourceMismatchDuringConfiguration() throws Exception { String agentName = "agent1"; String sourceType = "seq"; String channelType = "memory"; String sinkType = "avro"; Map<String, String> properties = getProperties(agentName, sourceType, channelType, sinkType); properties.put(agentName + ".channels.channel1.capacity", "1000"); properties.put(agentName + ".channels.channel1.transactionCapacity", "1000"); properties.put(agentName + ".sources.source1.batchSize", "1000"); properties.put(agentName + ".sinks.sink1.batch-size", "1000"); properties.put(agentName + ".sinks.sink1.hostname", "10.10.10.10"); properties.put(agentName + ".sinks.sink1.port", "1010"); MemoryConfigurationProvider provider = new MemoryConfigurationProvider(agentName, properties); MaterializedConfiguration config = provider.getConfiguration(); Assert.assertTrue(config.getSourceRunners().size() == 1); Assert.assertTrue(config.getChannels().size() == 1); Assert.assertTrue(config.getSinkRunners().size() == 1); properties.put(agentName + ".sources.source1.batchSize", "1001"); properties.put(agentName + ".sinks.sink1.batch-size", "1000"); provider = new MemoryConfigurationProvider(agentName, properties); config = provider.getConfiguration(); Assert.assertTrue(config.getSourceRunners().size() == 0); Assert.assertTrue(config.getChannels().size() == 1); Assert.assertTrue(config.getSinkRunners().size() == 1); properties.put(agentName + ".sources.source1.batchSize", "1000"); properties.put(agentName + ".sinks.sink1.batch-size", "1001"); provider = new MemoryConfigurationProvider(agentName, properties); config = provider.getConfiguration(); Assert.assertTrue(config.getSourceRunners().size() == 1); Assert.assertTrue(config.getChannels().size() == 1); Assert.assertTrue(config.getSinkRunners().size() == 0); properties.put(agentName + ".sources.source1.batchSize", "1001"); properties.put(agentName + ".sinks.sink1.batch-size", "1001"); provider = new MemoryConfigurationProvider(agentName, properties); config = provider.getConfiguration(); Assert.assertTrue(config.getSourceRunners().size() == 0); Assert.assertTrue(config.getChannels().size() == 0); Assert.assertTrue(config.getSinkRunners().size() == 0); }