You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by ar...@apache.org on 2012/02/15 06:49:53 UTC
svn commit: r1244368 - in
/incubator/flume/branches/flume-728/flume-ng-core/src:
main/java/org/apache/flume/channel/MultiplexingChannelSelector.java
test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java
Author: arvind
Date: Wed Feb 15 05:49:52 2012
New Revision: 1244368
URL: http://svn.apache.org/viewvc?rev=1244368&view=rev
Log:
FLUME-945. Add the ability to specify a default channel for multiplexing selector.
(Hari Shreedharan via Arvind Prabhakar)
Modified:
incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java
incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java
Modified: incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java?rev=1244368&r1=1244367&r2=1244368&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MultiplexingChannelSelector.java Wed Feb 15 05:49:52 2012
@@ -37,10 +37,10 @@ public class MultiplexingChannelSelector
public static final String DEFAULT_MULTIPLEX_HEADER =
"flume.selector.header";
public static final String CONFIG_PREFIX_MAPPING = "mapping";
-
+ public static final String CONFIG_PREFIX_DEFAULT = "default";
@SuppressWarnings("unused")
private static final Logger LOG = LoggerFactory
- .getLogger(MultiplexingChannelSelector.class);
+ .getLogger(MultiplexingChannelSelector.class);
private static final List<Channel> EMPTY_LIST =
Collections.emptyList();
@@ -49,17 +49,20 @@ public class MultiplexingChannelSelector
private Map<String, List<Channel>> channelMapping;
+ private List<Channel> defaultChannels;
@Override
public List<Channel> getRequiredChannels(Event event) {
String headerValue = event.getHeaders().get(headerName);
if (headerValue == null || headerValue.trim().length() == 0) {
- return EMPTY_LIST;
+ return defaultChannels;
}
List<Channel> channels = channelMapping.get(headerValue);
+ //This header value does not point to anything
+ //Return default channel(s) here.
if (channels == null) {
- channels = EMPTY_LIST;
+ channels = defaultChannels;
}
return channels;
@@ -75,30 +78,31 @@ public class MultiplexingChannelSelector
this.headerName = context.getString(CONFIG_MULTIPLEX_HEADER_NAME,
DEFAULT_MULTIPLEX_HEADER);
- Map<String, Object> mapConfig = context.getSubProperties(
- CONFIG_PREFIX_MAPPING + ".");
-
Map<String, Channel> channelNameMap = new HashMap<String, Channel>();
-
for (Channel ch : getAllChannels()) {
channelNameMap.put(ch.getName(), ch);
}
+ defaultChannels = getChannelListFromNames(
+ context.getString(CONFIG_PREFIX_DEFAULT),
+ channelNameMap);
+
+ if(defaultChannels.isEmpty()){
+ throw new FlumeException("Default channel list empty");
+ }
+
+ Map<String, Object> mapConfig = context.getSubProperties(
+ CONFIG_PREFIX_MAPPING + ".");
+
channelMapping = new HashMap<String, List<Channel>>();
for (String headerValue : mapConfig.keySet()) {
- String[] chNames = ((String) mapConfig.get(headerValue)).split(" ");
- List<Channel> configuredChannels = new ArrayList<Channel>();
- for (String name : chNames) {
- Channel ch = channelNameMap.get(name);
- if (ch != null) {
- configuredChannels.add(ch);
- } else {
- throw new FlumeException("Selector channel not found: "
- + name);
- }
- }
+ List<Channel> configuredChannels = getChannelListFromNames(
+ (String) mapConfig.get(headerValue),
+ channelNameMap);
+ //This should not go to default channel(s)
+ //because this seems to be a bad way to configure.
if (configuredChannels.size() == 0) {
throw new FlumeException("No channel configured for when "
+ "header value is: " + headerValue);
@@ -108,10 +112,27 @@ public class MultiplexingChannelSelector
throw new FlumeException("Selector channel configured twice");
}
}
+ //If no mapping is configured, it is ok.
+ //All events will go to the default channel(s).
- if (channelMapping.size() == 0) {
- throw new FlumeException("No mapping configured for selector");
+ }
+
+ //Given a list of channel names as space delimited string,
+ //returns list of channels.
+ private List<Channel> getChannelListFromNames(String channels,
+ Map<String, Channel> channelNameMap){
+ List<Channel> configuredChannels = new ArrayList<Channel>();
+ String[] chNames = channels.split(" ");
+ for (String name : chNames) {
+ Channel ch = channelNameMap.get(name);
+ if (ch != null) {
+ configuredChannels.add(ch);
+ } else {
+ throw new FlumeException("Selector channel not found: "
+ + name);
+ }
}
+ return configuredChannels;
}
}
Modified: incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java?rev=1244368&r1=1244367&r2=1244368&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestMultiplexingChannelSelector.java Wed Feb 15 05:49:52 2012
@@ -50,6 +50,7 @@ public class TestMultiplexingChannelSele
config.put("mapping.foo", "ch1 ch2");
config.put("mapping.bar", "ch2 ch3");
config.put("mapping.xyz", "ch1 ch2 ch3");
+ config.put("default", "ch1 ch3");
selector = ChannelSelectorFactory.create(channels, config);
}
@@ -96,6 +97,8 @@ public class TestMultiplexingChannelSele
Assert.assertTrue(optCh3.size() == 0);
}
+ //If the header information cannot map the event to any of the channels
+ //it should always be mapped to the default channel(s).
@Test
public void testNoSelection() throws Exception {
Assert.assertTrue(selector instanceof MultiplexingChannelSelector);
@@ -104,8 +107,9 @@ public class TestMultiplexingChannelSele
List<Channel> reqCh1 = selector.getRequiredChannels(noHeaderEvent);
List<Channel> optCh1 = selector.getOptionalChannels(noHeaderEvent);
-
- Assert.assertTrue(reqCh1.size() == 0);
+ Assert.assertEquals(2, reqCh1.size());
+ Assert.assertTrue(reqCh1.get(0).getName().equals("ch1"));
+ Assert.assertTrue(reqCh1.get(1).getName().equals("ch3"));
Assert.assertTrue(optCh1.size() == 0);
Map<String, String> header2 = new HashMap<String, String>();
@@ -115,8 +119,9 @@ public class TestMultiplexingChannelSele
List<Channel> reqCh2 = selector.getRequiredChannels(invalidHeaderEvent);
List<Channel> optCh2 = selector.getOptionalChannels(invalidHeaderEvent);
-
- Assert.assertTrue(reqCh2.size() == 0);
+ Assert.assertEquals(2, reqCh2.size());
+ Assert.assertTrue(reqCh2.get(0).getName().equals("ch1"));
+ Assert.assertTrue(reqCh2.get(1).getName().equals("ch3"));
Assert.assertTrue(optCh2.size() == 0);
Map<String, String> header3 = new HashMap<String, String>();
@@ -126,8 +131,9 @@ public class TestMultiplexingChannelSele
List<Channel> reqCh3 = selector.getRequiredChannels(unmatchedHeaderEvent);
List<Channel> optCh3 = selector.getOptionalChannels(unmatchedHeaderEvent);
-
- Assert.assertTrue(reqCh3.size() == 0);
+ Assert.assertEquals(2, reqCh3.size());
+ Assert.assertTrue(reqCh3.get(0).getName().equals("ch1"));
+ Assert.assertTrue(reqCh3.get(1).getName().equals("ch3"));
Assert.assertTrue(optCh3.size() == 0);