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:53 UTC
svn commit: r1338910 - in /incubator/flume/trunk/flume-ng-core/src:
main/java/org/apache/flume/interceptor/
test/java/org/apache/flume/interceptor/
Author: arvind
Date: Tue May 15 21:46:53 2012
New Revision: 1338910
URL: http://svn.apache.org/viewvc?rev=1338910&view=rev
Log:
FLUME-1157. Implement interceptors/decorators.
(Mike Percy via Arvind Prabhakar)
Added:
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java (with props)
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java (with props)
incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/
incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java (with props)
incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java (with props)
Added: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java?rev=1338910&view=auto
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java (added)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java Tue May 15 21:46:53 2012
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.interceptor;
+
+import java.util.List;
+import org.apache.flume.Event;
+import org.apache.flume.conf.Configurable;
+
+public interface Interceptor {
+ /**
+ * Any initialization / startup needed by the Interceptor.
+ */
+ public void initialize();
+
+ /**
+ * Interception of a single {@link Event}.
+ * @param event Event to be intercepted
+ * @return Original or modified event, or {@code null} if the Event
+ * is to be dropped (i.e. filtered out).
+ */
+ public Event intercept(Event event);
+
+ /**
+ * Interception of a batch of {@linkplain Event events}.
+ * @param events Input list of events
+ * @return Output list of events. The size of output list MUST NOT BE GREATER
+ * than the size of the input list (i.e. transformation and removal ONLY).
+ * Also, this method MUST NOT return {@code null}. If all events are dropped,
+ * then an empty List is returned.
+ */
+ public List<Event> intercept(List<Event> events);
+
+ /**
+ * Perform any closing / shutdown needed by the Interceptor.
+ */
+ public void close();
+
+ /** Builder implementations MUST have a no-arg constructor */
+ public interface Builder extends Configurable {
+ public Interceptor build();
+ }
+}
Propchange: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/Interceptor.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java?rev=1338910&view=auto
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java (added)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java Tue May 15 21:46:53 2012
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.interceptor;
+
+import com.google.common.base.Preconditions;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.flume.Event;
+import com.google.common.collect.Lists;
+
+/**
+ * Implementation of Interceptor that calls a list of other Interceptors
+ * serially.
+ */
+public class InterceptorChain implements Interceptor {
+
+ // list of interceptors that will be traversed, in order
+ private List<Interceptor> interceptors;
+
+ public InterceptorChain() {
+ interceptors = Lists.newLinkedList();
+ }
+
+ public void setInterceptors(List<Interceptor> interceptors) {
+ this.interceptors = interceptors;
+ }
+
+ @Override
+ public Event intercept(Event event) {
+ for (Interceptor interceptor : interceptors) {
+ if (event == null) {
+ return null;
+ }
+ event = interceptor.intercept(event);
+ }
+ return event;
+ }
+
+ @Override
+ public List<Event> intercept(List<Event> events) {
+ for (Interceptor interceptor : interceptors) {
+ if (events.isEmpty()) {
+ return events;
+ }
+ events = interceptor.intercept(events);
+ Preconditions.checkNotNull(events,
+ "Event list returned null from interceptor %s", interceptor);
+ }
+ return events;
+ }
+
+ @Override
+ public void initialize() {
+ Iterator<Interceptor> iter = interceptors.iterator();
+ while (iter.hasNext()) {
+ Interceptor interceptor = iter.next();
+ interceptor.initialize();
+ }
+ }
+
+ @Override
+ public void close() {
+ Iterator<Interceptor> iter = interceptors.iterator();
+ while (iter.hasNext()) {
+ Interceptor interceptor = iter.next();
+ interceptor.close();
+ }
+ }
+
+}
Propchange: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/interceptor/InterceptorChain.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java?rev=1338910&view=auto
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java (added)
+++ incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java Tue May 15 21:46:53 2012
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.flume.interceptor;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+
+/**
+ * Simple interceptor that removes a header called "Bad-Words" from all events.
+ */
+public class CensoringInterceptor implements Interceptor {
+
+ private CensoringInterceptor() {
+ // no-op
+ }
+
+ @Override
+ public void initialize() {
+ // no-op
+ }
+
+ @Override
+ public Event intercept(Event event) {
+ Map<String, String> headers = event.getHeaders();
+ if (headers.containsKey("Bad-Words")) {
+ headers.remove("Bad-Words");
+ }
+ return event;
+ }
+
+ @Override
+ public List<Event> intercept(List<Event> events) {
+ for (Event e : events) {
+ intercept(e);
+ }
+ return events;
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+
+ public static class Builder implements Interceptor.Builder {
+
+ @Override
+ public Interceptor build() {
+ return new CensoringInterceptor();
+ }
+
+ @Override
+ public void configure(Context context) {
+ // no-op
+ }
+
+ }
+
+}
Propchange: incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/CensoringInterceptor.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java?rev=1338910&view=auto
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java (added)
+++ incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java Tue May 15 21:46:53 2012
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.interceptor;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import java.util.Map;
+import org.apache.flume.Channel;
+import org.apache.flume.ChannelSelector;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.Transaction;
+import org.apache.flume.channel.ChannelProcessor;
+import org.apache.flume.channel.MemoryChannel;
+import org.apache.flume.channel.ReplicatingChannelSelector;
+import org.apache.flume.event.EventBuilder;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestCensoringInterceptor {
+
+ Logger logger =
+ LoggerFactory.getLogger(TestCensoringInterceptor.class);
+
+ @Test
+ public void testCensor() {
+
+ MemoryChannel memCh = new MemoryChannel();
+ memCh.configure(new Context());
+ memCh.start();
+
+ ChannelSelector cs = new ReplicatingChannelSelector();
+ cs.setChannels(Lists.<Channel>newArrayList(memCh));
+ ChannelProcessor cp = new ChannelProcessor(cs);
+
+ // source config
+ Map<String, String> cfgMap = Maps.newHashMap();
+ cfgMap.put("interceptors", "a");
+ String builderClass = CensoringInterceptor.Builder.class.getName();
+ cfgMap.put("interceptors.a.type", builderClass);
+ Context ctx = new Context(cfgMap);
+
+ // setup
+ cp.configure(ctx);
+ cp.initialize();
+
+ Map<String, String> headers = Maps.newHashMap();
+ String badWord = "scribe";
+ headers.put("Bad-Words", badWord);
+ Event event1 = EventBuilder.withBody("test", Charsets.UTF_8, headers);
+ Assert.assertEquals(badWord, event1.getHeaders().get("Bad-Words"));
+ cp.processEvent(event1);
+
+ Transaction tx = memCh.getTransaction();
+ tx.begin();
+
+ Event event1a = memCh.take();
+ Assert.assertNull(event1a.getHeaders().get("Bad-Words"));
+
+ tx.commit();
+ tx.close();
+
+ // cleanup / shutdown
+ cp.close();
+ memCh.stop();
+ }
+}
Propchange: incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestCensoringInterceptor.java
------------------------------------------------------------------------------
svn:eol-style = native