You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by mp...@apache.org on 2012/06/26 07:24:50 UTC
svn commit: r1353823 - in /incubator/flume/trunk:
flume-ng-channels/flume-recoverable-memory-channel/src/test/java/org/apache/flume/channel/recoverable/memory/
flume-ng-configuration/src/main/java/org/apache/flume/conf/sink/
flume-ng-core/src/main/java...
Author: mpercy
Date: Tue Jun 26 05:24:48 2012
New Revision: 1353823
URL: http://svn.apache.org/viewvc?rev=1353823&view=rev
Log:
FLUME-1315. Null sink should support batching.
(Hari Shreedharan via Mike Percy)
Modified:
incubator/flume/trunk/flume-ng-channels/flume-recoverable-memory-channel/src/test/java/org/apache/flume/channel/recoverable/memory/TestRecoverableMemoryChannel.java
incubator/flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/sink/SinkConfiguration.java
incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java
incubator/flume/trunk/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractLogicalNodeManager.java
Modified: incubator/flume/trunk/flume-ng-channels/flume-recoverable-memory-channel/src/test/java/org/apache/flume/channel/recoverable/memory/TestRecoverableMemoryChannel.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-channels/flume-recoverable-memory-channel/src/test/java/org/apache/flume/channel/recoverable/memory/TestRecoverableMemoryChannel.java?rev=1353823&r1=1353822&r2=1353823&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-channels/flume-recoverable-memory-channel/src/test/java/org/apache/flume/channel/recoverable/memory/TestRecoverableMemoryChannel.java (original)
+++ incubator/flume/trunk/flume-ng-channels/flume-recoverable-memory-channel/src/test/java/org/apache/flume/channel/recoverable/memory/TestRecoverableMemoryChannel.java Tue Jun 26 05:24:48 2012
@@ -116,8 +116,11 @@ public class TestRecoverableMemoryChanne
}
@Test
public void testRollbackWithSink() throws Exception {
- final NullSink sink = new NullSink();
- sink.setChannel(channel);
+ final NullSink nullSink = new NullSink();
+ Context ctx = new Context();
+ ctx.put("batchSize", "1");
+ nullSink.configure(ctx);
+ nullSink.setChannel(channel);
final int numItems = 99;
Thread t = new Thread() {
@Override
@@ -125,7 +128,7 @@ public class TestRecoverableMemoryChanne
int count = 0;
while(count++ < numItems) {
try {
- sink.process();
+ nullSink.process();
Thread.sleep(1);
} catch(EventDeliveryException e) {
break;
Modified: incubator/flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/sink/SinkConfiguration.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/sink/SinkConfiguration.java?rev=1353823&r1=1353822&r2=1353823&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/sink/SinkConfiguration.java (original)
+++ incubator/flume/trunk/flume-ng-configuration/src/main/java/org/apache/flume/conf/sink/SinkConfiguration.java Tue Jun 26 05:24:48 2012
@@ -80,7 +80,7 @@ public class SinkConfiguration extends C
*
* @see NullSink
*/
- NULL(null),
+ NULL("org.apache.flume.conf.sink.NullSinkConfiguration"),
/**
* Logger sink
Modified: incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java?rev=1353823&r1=1353822&r2=1353823&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java (original)
+++ incubator/flume/trunk/flume-ng-core/src/main/java/org/apache/flume/sink/NullSink.java Tue Jun 26 05:24:48 2012
@@ -17,12 +17,15 @@
*/
package org.apache.flume.sink;
+import com.google.common.base.Preconditions;
import org.apache.flume.Channel;
+import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Sink;
import org.apache.flume.Transaction;
+import org.apache.flume.conf.Configurable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,60 +47,83 @@ import org.slf4j.LoggerFactory;
* TODO
* </p>
*/
-public class NullSink extends AbstractSink {
+public class NullSink extends AbstractSink implements Configurable {
private static final Logger logger = LoggerFactory.getLogger(NullSink.class);
+ private static final int DFLT_BATCH_SIZE = 100;
+
private CounterGroup counterGroup;
+ private int batchSize = DFLT_BATCH_SIZE;
public NullSink() {
counterGroup = new CounterGroup();
}
@Override
+ public void configure(Context context) {
+ batchSize = context.getInteger("batchSize", DFLT_BATCH_SIZE);
+ logger.debug(this.getName() + " " +
+ "batch size set to " + String.valueOf(batchSize));
+ Preconditions.checkArgument(batchSize > 0, "Batch size must be > 0");
+ }
+
+ @Override
public Status process() throws EventDeliveryException {
+ Status status = Status.READY;
+
Channel channel = getChannel();
Transaction transaction = channel.getTransaction();
Event event = null;
try {
transaction.begin();
- event = channel.take();
- transaction.commit();
-
- if (event != null) {
- // logger.debug("Consumed the event: " + event);
- counterGroup.incrementAndGet("events.successful");
- return Status.READY;
- } else {
- return Status.BACKOFF;
+ int i = 0;
+ for (i = 0; i < batchSize; i++) {
+ event = channel.take();
+ if(event == null) {
+ status = Status.BACKOFF;
+ break;
+ }
}
+ transaction.commit();
+ counterGroup.addAndGet("events.success", (long) Math.min(batchSize, i));
+ counterGroup.incrementAndGet("transaction.success");
} catch (Exception ex) {
transaction.rollback();
- counterGroup.incrementAndGet("events.failed");
+ counterGroup.incrementAndGet("transaction.failed");
logger.error("Failed to deliver event. Exception follows.", ex);
throw new EventDeliveryException("Failed to deliver event: " + event, ex);
} finally {
transaction.close();
}
+
+ return status;
}
@Override
public void start() {
- logger.info("Null sink starting");
+ logger.info("Starting {}...", this);
+ counterGroup.setName(this.getName());
super.start();
- logger.debug("Null sink started");
+ logger.info("Null sink {} started.", getName());
}
@Override
public void stop() {
- logger.info("Null sink stopping");
+ logger.info("Null sink {} stopping...", getName());
super.stop();
- logger.info("Null sink stopped. Event metrics:{}", counterGroup);
+ logger.info("Null sink {} stopped. Event metrics: {}",
+ getName(), counterGroup);
+ }
+
+ @Override
+ public String toString() {
+ return "NullSink " + getName() + " { batchSize: " + batchSize + " }";
}
}
Modified: incubator/flume/trunk/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractLogicalNodeManager.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractLogicalNodeManager.java?rev=1353823&r1=1353822&r2=1353823&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractLogicalNodeManager.java (original)
+++ incubator/flume/trunk/flume-ng-node/src/test/java/org/apache/flume/node/TestAbstractLogicalNodeManager.java Tue Jun 26 05:24:48 2012
@@ -160,7 +160,8 @@ public class TestAbstractLogicalNodeMana
generatorSource.setChannelProcessor(new ChannelProcessor(rcs));
- Sink nullSink = new NullSink();
+ NullSink nullSink = new NullSink();
+ nullSink.configure(new Context());
nullSink.setChannel(channel);
nodeManager.add(SourceRunner.forSource(generatorSource));
@@ -200,7 +201,8 @@ public class TestAbstractLogicalNodeMana
source.setChannelProcessor(new ChannelProcessor(rcs));
- Sink sink = new NullSink();
+ NullSink sink = new NullSink();
+ sink.configure(new Context());
sink.setChannel(channel);
nodeManager.add(SourceRunner.forSource(source));