flume的memeryChannel中transactionCapacity和sink的batchsize需要注意事项

最近在做flume的实时日志收集，用flume默认的配置后，发现不是完全实时的，于是看了一下，原来是memeryChannel的transactionCapacity在作怪，因为他默认是100，也就是说收集端的sink会在收集到了100条以后再去提交事务（即发送到下一个目的地），于是我修改了transactionCapacity到10，想看看是不是会更加实时一点，结果发现收集日志的agent启动的时候报错了。

16/04/29 09:36:15 ERROR sink.AbstractRpcSink: Rpc Sink avro-sink: Unable to get event from channel memoryChannel. Exception follows.
org.apache.flume.ChannelException: Take list for MemoryTransaction, capacity 10 full, consider committing more frequently, increasing capacity, or increasing thread count
at org.apache.flume.channel.MemoryChannel$MemoryTransaction.doTake(MemoryChannel.java:96)
at org.apache.flume.channel.BasicTransactionSemantics.take(BasicTransactionSemantics.java:113)
at org.apache.flume.channel.BasicChannelSemantics.take(BasicChannelSemantics.java:95)
at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:354)
at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
at java.lang.Thread.run(Thread.java:745)

于是很纳闷，为什么默认值100可以，而设置10就会说小了呢，于是查阅资料，发现原来是sink的batchsize参数在作怪，下面，我就来理一理这个来龙去脉，这个sink的batchsize是什么意思呢，就是sink会一次从channel中取多少个event去发送，而这个发送是要最终以事务的形式去发送的，因此这个batchsize的event会传送到一个事务的缓存队列中（takeList），这是一个双向队列，这个队列可以在事务失败时进行回滚（也就是把取出来的数据吐memeryChannel的queue中），它的初始大小就是transactionCapacity定义的大小，源码中有： takeList = new LinkedBlockingDeque<Event>(transCapacity); 源码来自https://segmentfault.com/a/1190000003586635的分享。

再看这个错误抛出的地方：

 if(takeList.remainingCapacity() == 0) {
        throw new ChannelException("Take list for MemoryTransaction, capacity " +
            takeList.size() + " full, consider committing more frequently, " +
            "increasing capacity, or increasing thread count");
    }

在上面的情况中，sink一次取100个events，塞到takelist中，在塞了10个后，就会引发上述异常，因此，这个错误的解决办法就是：在sink中，channel的transactionCapacity参数不能小于sink的batchsize。有点不明白，这么明显的一个坑，apache为什么不去补一下呢？