You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by wi...@apache.org on 2012/07/28 03:46:02 UTC
svn commit: r1366581 -
/flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java
Author: will
Date: Sat Jul 28 01:46:01 2012
New Revision: 1366581
URL: http://svn.apache.org/viewvc?rev=1366581&view=rev
Log:
FLUME-1392. Inactive channels get added to source channels list causing NPE.
(Hari Shreedharan via Will McQueen)
Modified:
flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java
Modified: flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java
URL: http://svn.apache.org/viewvc/flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java?rev=1366581&r1=1366580&r2=1366581&view=diff
==============================================================================
--- flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java (original)
+++ flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/FlumeConfiguration.java Sat Jul 28 01:46:01 2012
@@ -551,6 +551,12 @@ public class FlumeConfiguration {
channels.addAll(srcConf.getChannels());
}
channels.retainAll(channelSet);
+ if(channels.isEmpty()){
+ throw new ConfigurationException(
+ "No Channels configured for " + sourceName);
+ }
+ srcContext.put(BasicConfigurationConstants.CONFIG_CHANNELS,
+ this.getSpaceDelimitedList(channels));
}
if ((configSpecified && srcConf.isNotFoundConfigClass()) ||
!configSpecified) {
@@ -655,6 +661,10 @@ public class FlumeConfiguration {
sinkConf.configure(sinkContext);
}
+ if(!channelSet.contains(sinkConf.getChannel())){
+ throw new ConfigurationException("Channel " +
+ sinkConf.getChannel() + " not in active set.");
+ }
if ((configSpecified && sinkConf.isNotFoundConfigClass()) ||
!configSpecified) {
newContextMap.put(sinkName, sinkContext);
@@ -665,8 +675,8 @@ public class FlumeConfiguration {
} catch (ConfigurationException e) {
iter.remove();
if (sinkConf != null) errorList.addAll(sinkConf.getErrors());
- logger.warn("Configuration empty for: " + sinkName + ".Removed.");
-
+ logger.warn("Configuration for : " + sinkName
+ + " has errors, and will be removed: ", e);
}
}
// Filter out any sinks that have invalid channel