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/05/15 23:46:02 UTC
svn commit: r1338909 - in /incubator/flume/trunk:
flume-ng-core/src/main/java/org/apache/flume/channel/
flume-ng-core/src/main/java/org/apache/flume/source/
flume-ng-core/src/test/java/org/apache/flume/source/
flume-ng-node/src/main/java/org/apache/flu...
Author: arvind
Date: Tue May 15 21:46:01 2012
New Revision: 1338909
URL: http://svn.apache.org/viewvc?rev=1338909&view=rev
Log:
FLUME-1157. Implement interceptors/decorators.
(Mike Percy via Arvind Prabhakar)
Modified:
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/EventDrivenSourceRunner.java
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/PollableSourceRunner.java
incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java
incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
Modified: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java?rev=1338909&r1=1338908&r2=1338909&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/channel/ChannelProcessor.java Tue May 15 21:46:01 2012
@@ -18,6 +18,8 @@
*/
package org.apache.flume.channel;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
@@ -26,8 +28,13 @@ import java.util.Map;
import org.apache.flume.Channel;
import org.apache.flume.ChannelException;
import org.apache.flume.ChannelSelector;
+import org.apache.flume.Context;
import org.apache.flume.Event;
+import org.apache.flume.FlumeException;
+import org.apache.flume.interceptor.Interceptor;
+import org.apache.flume.interceptor.InterceptorChain;
import org.apache.flume.Transaction;
+import org.apache.flume.conf.Configurable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,15 +49,77 @@ import org.slf4j.LoggerFactory;
* channels are
* {@linkplain ChannelSelector#getOptionalChannels(Event) optional}.
*/
-public class ChannelProcessor {
+public class ChannelProcessor implements Configurable {
private static final Logger LOG = LoggerFactory.getLogger(
ChannelProcessor.class);
private final ChannelSelector selector;
+ private final InterceptorChain interceptorChain;
public ChannelProcessor(ChannelSelector selector) {
this.selector = selector;
+ this.interceptorChain = new InterceptorChain();
+ }
+
+ public void initialize() {
+ interceptorChain.initialize();
+ }
+
+ public void close() {
+ interceptorChain.close();
+ }
+
+ /**
+ * The Context of the associated Source is passed.
+ * @param context
+ */
+ @Override
+ public void configure(Context context) {
+ configureInterceptors(context);
+ }
+
+ // WARNING: throws FlumeException (is that ok?)
+ private void configureInterceptors(Context context) {
+
+ List<Interceptor> interceptors = Lists.newLinkedList();
+
+ String interceptorListStr = context.getString("interceptors", "");
+ if (interceptorListStr.isEmpty()) {
+ return;
+ }
+ String[] interceptorNames = interceptorListStr.split("\\s+");
+
+ Context interceptorContexts =
+ new Context(context.getSubProperties("interceptors."));
+
+ // run through and instantiate all the interceptors specified in the Context
+ for (String interceptorName : interceptorNames) {
+ Context interceptorContext = new Context(
+ interceptorContexts.getSubProperties(interceptorName + "."));
+ String type = interceptorContext.getString("type");
+ try {
+ // TODO: support aliases for built-in interceptor types
+ Class<?> clazz = Class.forName(type);
+ if (Interceptor.Builder.class.isAssignableFrom(clazz)) {
+ Interceptor.Builder builder =
+ (Interceptor.Builder) clazz.newInstance();
+ builder.configure(interceptorContext);
+ interceptors.add(builder.build());
+ }
+ } catch (ClassNotFoundException e) {
+ LOG.error("Builder class not found. Exception follows.", e);
+ throw new FlumeException("Interceptor.Builder not found.", e);
+ } catch (InstantiationException e) {
+ LOG.error("Could not instantiate Builder. Exception follows.", e);
+ throw new FlumeException("Interceptor.Builder not constructable.", e);
+ } catch (IllegalAccessException e) {
+ LOG.error("Unable to access Builder. Exception follows.", e);
+ throw new FlumeException("Unable to access Interceptor.Builder.", e);
+ }
+ }
+
+ interceptorChain.setInterceptors(interceptors);
}
public ChannelSelector getSelector() {
@@ -70,6 +139,10 @@ public class ChannelProcessor {
* @throws ChannelException when a write to a required channel fails.
*/
public void processEventBatch(List<Event> events) {
+ Preconditions.checkNotNull(events, "Event list must not be null");
+
+ events = interceptorChain.intercept(events);
+
Map<Channel, List<Event>> reqChannelQueue =
new LinkedHashMap<Channel, List<Event>>();
@@ -102,7 +175,6 @@ public class ChannelProcessor {
}
// Process required channels
-
for (Channel reqChannel : reqChannelQueue.keySet()) {
Transaction tx = null;
try {
@@ -165,15 +237,20 @@ public class ChannelProcessor {
*/
public void processEvent(Event event) {
+ event = interceptorChain.intercept(event);
+ if (event == null) {
+ return;
+ }
+
// Process required channels
List<Channel> requiredChannels = selector.getRequiredChannels(event);
- for (Channel requriedChannel : requiredChannels) {
- Transaction tx = null;
+ for (Channel requiredChannel : requiredChannels) {
+ Transaction tx = requiredChannel.getTransaction();
+ Preconditions.checkNotNull(tx, "Transaction object must not be null");
try {
- tx = requriedChannel.getTransaction();
tx.begin();
- requriedChannel.put(event);
+ requiredChannel.put(event);
tx.commit();
} catch (ChannelException ex) {
Modified: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/EventDrivenSourceRunner.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/EventDrivenSourceRunner.java?rev=1338909&r1=1338908&r2=1338909&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/EventDrivenSourceRunner.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/EventDrivenSourceRunner.java Tue May 15 21:46:01 2012
@@ -19,7 +19,9 @@
package org.apache.flume.source;
+import org.apache.flume.Source;
import org.apache.flume.SourceRunner;
+import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.lifecycle.LifecycleState;
/**
@@ -36,13 +38,19 @@ public class EventDrivenSourceRunner ext
@Override
public void start() {
- getSource().start();
+ Source source = getSource();
+ ChannelProcessor cp = source.getChannelProcessor();
+ cp.initialize();
+ source.start();
lifecycleState = LifecycleState.START;
}
@Override
public void stop() {
- getSource().stop();
+ Source source = getSource();
+ source.stop();
+ ChannelProcessor cp = source.getChannelProcessor();
+ cp.close();
lifecycleState = LifecycleState.STOP;
}
Modified: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/PollableSourceRunner.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/PollableSourceRunner.java?rev=1338909&r1=1338908&r2=1338909&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/PollableSourceRunner.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/PollableSourceRunner.java Tue May 15 21:46:01 2012
@@ -24,7 +24,9 @@ import java.util.concurrent.atomic.Atomi
import org.apache.flume.CounterGroup;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
+import org.apache.flume.Source;
import org.apache.flume.SourceRunner;
+import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.lifecycle.LifecycleState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,7 +69,8 @@ public class PollableSourceRunner extend
@Override
public void start() {
PollableSource source = (PollableSource) getSource();
-
+ ChannelProcessor cp = source.getChannelProcessor();
+ cp.initialize();
source.start();
runner = new PollingRunner();
@@ -98,7 +101,10 @@ public class PollableSourceRunner extend
Thread.currentThread().interrupt();
}
- getSource().stop();
+ Source source = getSource();
+ source.stop();
+ ChannelProcessor cp = source.getChannelProcessor();
+ cp.close();
lifecycleState = LifecycleState.STOP;
}
Modified: incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java?rev=1338909&r1=1338908&r2=1338909&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestPollableSourceRunner.java Tue May 15 21:46:01 2012
@@ -19,16 +19,20 @@
package org.apache.flume.source;
+import com.google.common.collect.Lists;
import java.util.concurrent.CountDownLatch;
import org.apache.flume.Channel;
+import org.apache.flume.ChannelSelector;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.Transaction;
import org.apache.flume.channel.ChannelProcessor;
+import org.apache.flume.channel.ChannelSelectorFactory;
import org.apache.flume.channel.MemoryChannel;
+import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.lifecycle.LifecycleState;
@@ -56,9 +60,13 @@ public class TestPollableSourceRunner {
Configurables.configure(channel, new Context());
+ final ChannelSelector cs = new ReplicatingChannelSelector();
+ cs.setChannels(Lists.newArrayList(channel));
+
PollableSource source = new PollableSource() {
private String name;
+ private ChannelProcessor cp = new ChannelProcessor(cs);
@Override
public Status process() throws EventDeliveryException {
@@ -114,13 +122,12 @@ public class TestPollableSourceRunner {
@Override
public void setChannelProcessor(ChannelProcessor channelProcessor) {
- // Not used
+ cp = channelProcessor;
}
@Override
public ChannelProcessor getChannelProcessor() {
- // Not used
- return null;
+ return cp;
}
};
Modified: incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java?rev=1338909&r1=1338908&r2=1338909&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java (original)
+++ incubator/flume/trunk/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java Tue May 15 21:46:01 2012
@@ -291,6 +291,7 @@ public class PropertiesFileConfiguration
channels, selectorConfig);
ChannelProcessor channelProcessor = new ChannelProcessor(selector);
+ Configurables.configure(channelProcessor, config);
source.setChannelProcessor(channelProcessor);
conf.getSourceRunners().put(comp.getComponentName(),
@@ -318,6 +319,7 @@ public class PropertiesFileConfiguration
channels, selectorConfig);
ChannelProcessor channelProcessor = new ChannelProcessor(selector);
+ Configurables.configure(channelProcessor, context);
source.setChannelProcessor(channelProcessor);
conf.getSourceRunners().put(src,