svn commit: r1311535 - in /incubator/flume/trunk/flume-ng-core/src: main/java/org/apache/flume/source/AvroSource.java test/java/org/apache/flume/source/TestAvroSource.java

[br...@apache.org]

Author: brock
Date: Tue Apr 10 00:47:22 2012
New Revision: 1311535

URL: http://svn.apache.org/viewvc?rev=1311535&view=rev
Log:
FLUME-1011: AvroSource should have a configurable max thread count

(Juhani Connolly via Brock Noland)

Modified:
    incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/AvroSource.java
    incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestAvroSource.java

Modified: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/AvroSource.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/AvroSource.java?rev=1311535&r1=1311534&r2=1311535&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/AvroSource.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/source/AvroSource.java Tue Apr 10 00:47:22 2012
@@ -24,6 +24,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executors;
 
 import org.apache.avro.ipc.NettyServer;
 import org.apache.avro.ipc.Responder;
@@ -41,6 +42,7 @@ import org.apache.flume.event.EventBuild
 import org.apache.flume.source.avro.AvroFlumeEvent;
 import org.apache.flume.source.avro.AvroSourceProtocol;
 import org.apache.flume.source.avro.Status;
+import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -85,6 +87,12 @@ import org.slf4j.LoggerFactory;
  * <td>TCP port / int</td>
  * <td>none (required)</td>
  * </tr>
+ * <tr>
+ * <td><tt>threads</tt></td>
+ * <td>Max number of threads assigned to thread pool, 0 being unlimited</td>
+ * <td>Count / int</td>
+ * <td>0(optional)</td>
+ * </tr>
  * </table>
  * <p>
  * <b>Metrics</b>
@@ -96,6 +104,8 @@ import org.slf4j.LoggerFactory;
 public class AvroSource extends AbstractSource implements EventDrivenSource,
     Configurable, AvroSourceProtocol {
 
+  private static final String THREADS = "threads";
+
   private static final Logger logger = LoggerFactory
       .getLogger(AvroSource.class);
 
@@ -105,6 +115,8 @@ public class AvroSource extends Abstract
   private Server server;
   private CounterGroup counterGroup;
 
+  private int maxThreads;
+
   public AvroSource() {
     counterGroup = new CounterGroup();
   }
@@ -113,6 +125,12 @@ public class AvroSource extends Abstract
   public void configure(Context context) {
     port = Integer.parseInt(context.getString("port"));
     bindAddress = context.getString("bind");
+    try {
+      maxThreads = context.getInteger(THREADS, 0);
+    } catch (NumberFormatException e) {
+      logger.warn("AVRO source\'s \"threads\" property must specify an integer value.",
+              context.getString(THREADS));
+    }
   }
 
   @Override
@@ -120,8 +138,14 @@ public class AvroSource extends Abstract
     logger.info("Avro source starting:{}", this);
 
     Responder responder = new SpecificResponder(AvroSourceProtocol.class, this);
-    server = new NettyServer(responder,
-        new InetSocketAddress(bindAddress, port));
+    if(maxThreads <= 0) {
+      server = new NettyServer(responder,
+              new InetSocketAddress(bindAddress, port));
+    } else {
+      server = new NettyServer(responder, new InetSocketAddress(bindAddress, port),
+              new NioServerSocketChannelFactory(Executors.newFixedThreadPool(maxThreads),
+                      Executors.newFixedThreadPool(maxThreads)));
+    }
 
     server.start();
 

Modified: incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestAvroSource.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestAvroSource.java?rev=1311535&r1=1311534&r2=1311535&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestAvroSource.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/test/java/org/apache/flume/source/TestAvroSource.java Tue Apr 10 00:47:22 2012
@@ -120,6 +120,7 @@ public class TestAvroSource {
 
         context.put("port", String.valueOf(selectedPort = 41414 + i));
         context.put("bind", "0.0.0.0");
+        context.put("threads", "50");
 
         Configurables.configure(source, context);