【发布时间】:2015-10-16 17:09:04
【问题描述】:
我正在创建一个 Flume 代理的概念验证,该代理将缓冲事件并在接收器不可用时停止消耗来自源的事件。只有当 sink 再次可用时,才应该处理缓冲的事件,然后 source 重新开始消费。
为此,我创建了一个简单的代理,它从 SpoolDir 读取并写入文件。为了模拟接收器服务已关闭,我更改了文件权限,因此 Flume 无法对其进行写入。然后我启动 Flume 一些事件被缓冲在内存通道中,当通道容量已满时它会停止消耗事件,正如预期的那样。一旦文件变得可写,sink 就能够处理事件并且 Flume 恢复。但是,这只适用于不超过交易容量的情况。一旦超过事务容量,Flume 就永远不会恢复并不断写入以下错误:
2015-10-02 14:52:51,940 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR -
org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:160)] Unable to
deliver event. Exception follows.
org.apache.flume.EventDeliveryException: Failed to process transaction
at org.apache.flume.sink.RollingFileSink.process(RollingFileSink.java:218)
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)
Caused by: org.apache.flume.ChannelException: Take list for MemoryTransaction,
capacity 4 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.RollingFileSink.process(RollingFileSink.java:191)
... 3 more
一旦内存中缓冲的事件数超过事务容量 (4),就会发生此错误。不明白为什么,因为fileout的batchSize是1,所以应该一个一个的取出事件。
这是我正在使用的配置:
agent.sources = spool-src
agent.channels = mem-channel
agent.sinks = fileout
agent.sources.spool-src.channels = mem-channel
agent.sources.spool-src.type = spooldir
agent.sources.spool-src.spoolDir = /tmp/flume-spool
agent.sources.spool-src.batchSize = 1
agent.channels.mem-channel.type = memory
agent.channels.mem-channel.capacity = 10
agent.channels.mem-channel.transactionCapacity = 4
agent.sinks.fileout.channel = mem-channel
agent.sinks.fileout.type = file_roll
agent.sinks.fileout.sink.directory = /tmp/flume-output
agent.sinks.fileout.sink.rollInterval = 0
agent.sinks.fileout.batchSize = 1
我已经使用不同的通道容量和事务容量值(例如 3 和 3)测试了此配置,但没有发现通道容量已满且 Flume 能够恢复的情况。
【问题讨论】:
标签: flume