You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by pr...@apache.org on 2011/12/01 01:35:36 UTC
svn commit: r1208911 - in /incubator/flume/branches/flume-728:
flume-ng-core/src/main/java/org/apache/flume/
flume-ng-core/src/main/java/org/apache/flume/channel/
flume-ng-core/src/test/java/org/apache/flume/channel/
flume-ng-node/src/main/java/org/apa...
Author: prasadm
Date: Thu Dec 1 00:35:34 2011
New Revision: 1208911
URL: http://svn.apache.org/viewvc?rev=1208911&view=rev
Log:
Flume-814. Fanout channel support
Added:
incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/FanoutChannel.java
incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestFanoutChannel.java
Modified:
incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java
incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java
incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
Modified: incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java?rev=1208911&r1=1208910&r2=1208911&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/ChannelFactory.java Thu Dec 1 00:35:34 2011
@@ -17,6 +17,7 @@
*/
package org.apache.flume;
+import java.util.Map;
import java.util.Set;
public interface ChannelFactory {
@@ -27,6 +28,7 @@ public interface ChannelFactory {
public Channel create(String name) throws InstantiationException;
- public Set<String> getChannelNames();
+ public Channel createFanout(String chList, Map<String, Channel> chMap) throws InstantiationException;
+ public Set<String> getChannelNames();
}
Modified: incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java?rev=1208911&r1=1208910&r2=1208911&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/DefaultChannelFactory.java Thu Dec 1 00:35:34 2011
@@ -22,6 +22,7 @@ package org.apache.flume.channel;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import java.util.StringTokenizer;
import org.apache.flume.Channel;
import org.apache.flume.ChannelFactory;
@@ -101,4 +102,17 @@ public class DefaultChannelFactory imple
this.channelRegistry = channelRegistry;
}
+ // build a fanout channel from the list of channel names and map of <name,channels>
+ @Override
+ public Channel createFanout(String chList, Map<String, Channel> chMap)
+ throws InstantiationException {
+
+ FanoutChannel fnc = (FanoutChannel)create("fanout");
+ StringTokenizer tk = new StringTokenizer(chList, ",");
+ while (tk.hasMoreTokens()) {
+ fnc.addFanout(chMap.get(tk.nextToken()));
+ }
+ return fnc;
+ }
+
}
Added: incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/FanoutChannel.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/FanoutChannel.java?rev=1208911&view=auto
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/FanoutChannel.java (added)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/FanoutChannel.java Thu Dec 1 00:35:34 2011
@@ -0,0 +1,151 @@
+package org.apache.flume.channel;
+
+import java.util.LinkedList;
+
+import org.apache.flume.Channel;
+import org.apache.flume.ChannelException;
+import org.apache.flume.Event;
+import org.apache.flume.Transaction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FanoutChannel implements Channel {
+ private final Logger logger = LoggerFactory
+ .getLogger(FanoutChannel.class);
+
+ /**
+ * A wrapper transaction that does the operation on all channels.
+ * Note that there's no two phase commit. If one of the channels
+ * throws an exception, we still execute the operations for the
+ * rest. All the failed commits are rolled back at the end to
+ * maintain consistent transaction state.
+ * Note that the currently commit and rollback are not throwing
+ * exceptions * even if one of the underlying transactions fail.
+ */
+ public class wrapperTransaction implements Transaction {
+ @Override
+ public void begin() {
+ boolean txnErrors = false;
+ for (Channel ch : channelList) {
+ try {
+ ch.getTransaction().begin();
+ } catch (ChannelException e) {
+ txnErrors = true;
+ logger.warn("Error in fanout commit" + ch.getName(), e);
+ }
+ }
+ if (txnErrors == true) {
+ throw new ChannelException("Errors in fanout begin");
+ }
+ }
+
+ @Override
+ public void commit() {
+ boolean txnErrors = false;
+ LinkedList<Transaction> failedTxnList = new LinkedList<Transaction> ();
+ Transaction lastTxn = null;
+
+ for (Channel ch : channelList) {
+ try {
+ lastTxn = ch.getTransaction();
+ lastTxn.commit();
+ } catch (ChannelException e) {
+ txnErrors = true;
+ logger.warn("Error in fanout commit" + ch.getName(), e);
+ if (lastTxn != null) {
+ failedTxnList.add(lastTxn);
+ }
+ }
+ }
+ if (txnErrors == true) {
+ // TODO : Need some way to notify the caller about failed commits
+
+ // rollback the transactions that we couldn't successfully commit
+ for (Transaction tx : failedTxnList) {
+ try {
+ tx.rollback();
+ } catch (ChannelException e) {
+ // Ignore the exception during cleanup
+ }
+ }
+ }
+ }
+
+ @Override
+ public void rollback() {
+ boolean txnErrors = false;
+
+ for (Channel ch : channelList) {
+ try {
+ ch.getTransaction().rollback();
+ } catch (ChannelException e) {
+ txnErrors = true;
+ logger.warn("Error in fanout rollback" + ch.getName(), e);
+ }
+ }
+ if (txnErrors == true) {
+ // TODO : Need some way to notify the caller about failed rollbacks
+ }
+ }
+
+ @Override
+ public void close() {
+ boolean txnErrors = false;
+
+ for (Channel ch : channelList) {
+ try {
+ ch.getTransaction().close();
+ } catch (ChannelException e) {
+ txnErrors = true;
+ logger.warn("Error in fanout close" + ch.getName(), e);
+ }
+ }
+ if (txnErrors == true) {
+ throw new ChannelException("Errors in fanout close");
+ }
+ }
+ }
+
+ private LinkedList<Channel> channelList;
+ private wrapperTransaction txn;
+
+ public FanoutChannel() {
+ channelList = new LinkedList<Channel>();
+ txn = new wrapperTransaction();
+ }
+
+ public void addFanout(Channel ch) {
+ channelList.add(ch);
+ }
+
+ @Override
+ public void put(Event event) throws ChannelException {
+ for (Channel ch : channelList) {
+ ch.put(event);
+ }
+ }
+
+ @Override
+ public Event take() throws ChannelException {
+ // fanout is really for sources, we don't want to support take
+ throw new ChannelException("Can't take from fanout channel");
+ }
+
+ @Override
+ public Transaction getTransaction() {
+ return txn;
+ }
+
+ @Override
+ public void shutdown() {
+ for (Channel ch : channelList) {
+ ch.shutdown();
+ }
+ }
+
+ @Override
+ public String getName() {
+ return "SourceFanout";
+ }
+
+}
Modified: incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java?rev=1208911&r1=1208910&r2=1208911&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/main/java/org/apache/flume/channel/MemoryChannel.java Thu Dec 1 00:35:34 2011
@@ -255,7 +255,8 @@ public class MemoryChannel implements Ch
int myStamp = currentStamp.getAndIncrement();
StampedEvent stampedEvent = new StampedEvent(myStamp, event);
- queue.put(stampedEvent);
+ if (queue.offer(stampedEvent,keepAlive, TimeUnit.SECONDS) == false)
+ throw new ChannelException("put(" + event + ") timed out");
myTxn.logPut(stampedEvent, myStamp);
} catch (InterruptedException ex) {
Added: incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestFanoutChannel.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestFanoutChannel.java?rev=1208911&view=auto
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestFanoutChannel.java (added)
+++ incubator/flume/branches/flume-728/flume-ng-core/src/test/java/org/apache/flume/channel/TestFanoutChannel.java Thu Dec 1 00:35:34 2011
@@ -0,0 +1,132 @@
+package org.apache.flume.channel;
+
+import org.apache.flume.Channel;
+import org.apache.flume.ChannelException;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.Transaction;
+import org.apache.flume.channel.MemoryChannel;
+import org.apache.flume.channel.FanoutChannel.wrapperTransaction;
+import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.EventBuilder;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TestFanoutChannel {
+
+ private final Logger logger = LoggerFactory.getLogger(TestFanoutChannel.class);
+
+ @Before
+ public void setUp() {
+ }
+
+ @Test
+ public void testCommit() throws InterruptedException, EventDeliveryException {
+
+ Channel ch1 = getMemChannel(20);
+ Channel ch2 = getMemChannel(20);
+ FanoutChannel chFan = new FanoutChannel();
+ chFan.addFanout(ch1);
+ chFan.addFanout(ch2);
+
+ // write data into fanout channel and commit
+ putIntoFanoutChannel(chFan, true);
+ // verify the data in each channel
+ verifyChannelData(ch1);
+ verifyChannelData(ch2);
+
+ // write data into fanout channel and rollback
+ putIntoFanoutChannel(chFan, false);
+ // verify that channels have no events
+ verifyChannelEmpty(ch1);
+ verifyChannelEmpty(ch2);
+ }
+
+ @Test
+ public void testPutErrors() throws InterruptedException, EventDeliveryException {
+
+ Channel ch1 = getMemChannel(20);
+ Channel ch2 = getMemChannel(5); // smaller capacity queue
+ FanoutChannel chFan = new FanoutChannel();
+ chFan.addFanout(ch1);
+ chFan.addFanout(ch2);
+
+ // write data into fanout channel and commit
+ // it should run into error due to small channel overflow
+ try {
+ putIntoFanoutChannel(chFan, true);
+ } catch (ChannelException e) {
+ logger.warn("small channel overflow", e);
+ }
+
+ //verify that channels are empty as we rolled back due to error
+ verifyChannelEmpty(ch1);
+ verifyChannelEmpty(ch2);
+ }
+
+ private Channel getMemChannel(int Capacity) {
+ Channel ch1 = new MemoryChannel();
+ Context context1 = new Context();
+ context1.put("keep-alive", "1");
+ context1.put("capacity", Integer.toString(Capacity));
+ Configurables.configure(ch1, context1);
+ return ch1;
+ }
+
+ private void putIntoFanoutChannel(Channel chFan, boolean commitData) {
+ Event event;
+ int putCounter = 0;
+
+ Transaction transaction = chFan.getTransaction();
+ Assert.assertNotNull(transaction);
+
+ try {
+ transaction.begin();
+ for (putCounter = 0; putCounter < 10; putCounter++) {
+ event = EventBuilder.withBody(("test event" + putCounter).getBytes());
+ chFan.put(event);
+ }
+ if (commitData == true)
+ transaction.commit();
+ else
+ transaction.rollback();
+ } finally {
+ transaction.close();
+ }
+ }
+
+ private void verifyChannelData(Channel ch) {
+ Event event2;
+
+ Transaction transaction = ch.getTransaction();
+ Assert.assertNotNull(transaction);
+
+ transaction.begin();
+ for (int i = 0; i < 10; i++) {
+ event2 = ch.take();
+ Assert.assertNotNull("lost an event", event2);
+ Assert.assertArrayEquals(event2.getBody(), ("test event" + i).getBytes());
+ }
+ event2 = ch.take();
+ Assert.assertNull("extra event found", event2);
+ transaction.commit();
+ transaction.close();
+ }
+
+ private void verifyChannelEmpty(Channel ch) {
+ Event event2;
+
+ Transaction transaction = ch.getTransaction();
+ Assert.assertNotNull(transaction);
+
+ transaction.begin();
+ event2 = ch.take();
+ Assert.assertNull("extra event found", event2);
+ transaction.commit();
+ transaction.close();
+ }
+}
Modified: incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java?rev=1208911&r1=1208910&r2=1208911&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/conf/properties/PropertiesFileConfigurationProvider.java Thu Dec 1 00:35:34 2011
@@ -23,6 +23,7 @@ import java.io.FileReader;
import java.io.IOException;
import java.util.Map.Entry;
import java.util.Properties;
+import java.util.StringTokenizer;
import org.apache.flume.Channel;
import org.apache.flume.Context;
@@ -30,6 +31,7 @@ import org.apache.flume.Sink;
import org.apache.flume.SinkRunner;
import org.apache.flume.Source;
import org.apache.flume.SourceRunner;
+import org.apache.flume.channel.FanoutChannel;
import org.apache.flume.conf.Configurables;
import org.apache.flume.conf.file.AbstractFileConfigurationProvider;
import org.apache.flume.conf.file.SimpleNodeConfiguration;
@@ -241,8 +243,14 @@ public class PropertiesFileConfiguration
Configurables.configure(source, context);
- source.setChannel(conf.getChannels().get(
- comp.getConfiguration().get("channels")));
+ if (comp.getConfiguration().get("channels").contains(",")) {
+ // then create a fanout if there are channel for this source
+ source.setChannel(getChannelFactory().createFanout(
+ comp.getConfiguration().get("channels"),conf.getChannels()));
+ } else {
+ source.setChannel(conf.getChannels().get(
+ comp.getConfiguration().get("channels")));
+ }
conf.getSourceRunners().put(comp.getComponentName(),
SourceRunner.forSource(source));
}
Modified: incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java
URL: http://svn.apache.org/viewvc/incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java?rev=1208911&r1=1208910&r2=1208911&view=diff
==============================================================================
--- incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java (original)
+++ incubator/flume/branches/flume-728/flume-ng-node/src/main/java/org/apache/flume/node/Application.java Thu Dec 1 00:35:34 2011
@@ -34,6 +34,7 @@ import org.apache.flume.Constants;
import org.apache.flume.SinkFactory;
import org.apache.flume.SourceFactory;
import org.apache.flume.channel.DefaultChannelFactory;
+import org.apache.flume.channel.FanoutChannel;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.jdbc.JdbcChannel;
import org.apache.flume.conf.file.AbstractFileConfigurationProvider;
@@ -99,6 +100,7 @@ public class Application {
public void loadPlugins() {
channelFactory.register("memory", MemoryChannel.class);
channelFactory.register("jdbc", JdbcChannel.class);
+ channelFactory.register("fanout", FanoutChannel.class);
sourceFactory.register("seq", SequenceGeneratorSource.class);
sourceFactory.register("netcat", NetcatSource.class);