错误flume.SinkRunner：无法传递事件。例外如下

Question

I have one flume agent with one exec source, two custom channels and two hdfs sinks. while starting the agent, it gives me an error saying ERROR flume.SinkRunner: Unable to deliver event. Exception follows.java.lang.IllegalStateException: Channel closed.

我的配置文件是

agent.sources = source
agent.sinks = sink1 sink2
agent.channels = channel1 channel2

agent.sources.source.type = exec
agent.sources.source.command = cat /home/Flume/XMLFiles/
agent.sources.source.channels = channel1 channel2
agent.sources.source.selector.type = replicating

agent.channels.channel1.type = com.flume.Channel1
agent.channels.channel1.type = file
agent.channels.channel1.transactionCapacity = 100000
agent.channels.channel1.checkpointInterval = 3000

agent.channels.channel2.type = com.flume.Channel2
agent.channels.channel2.type = file
agent.channels.channel2.transactionCapacity = 100000
agent.channels.channel2.checkpointInterval = 3000

agent.sinks.sink1.type = hdfs
agent.sinks.sink1.hdfs.path = hdfs://ip:port/user/Channel1
agent.sinks.sink1.channel = channel1
agent.sinks.sink1.hdfs.filePrefix = test
agent.sinks.sink1.hdfs.fileSuffix = .log
agent.sinks.sink1.hdfs.rollInterval = 1200
agent.sinks.sink1.hdfs.batchSize = 1000
agent.sinks.sink1.hdfs.fileType = DataStream

agent.sinks.sink2.type = hdfs
agent.sinks.sink2.hdfs.path = hdfs://ip:port/user/Channel2
agent.sinks.sink2.channel = channel2
agent.sinks.sink2.hdfs.filePrefix = test
agent.sinks.sink2.hdfs.fileSuffix = .log
agent.sinks.sink2.hdfs.rollInterval = 1200
agent.sinks.sink2.hdfs.batchSize = 1000
agent.sinks.sink2.hdfs.fileType = DataStream

agent.sources.source.channels = channel1 channel2
agent.sinks.sink1.channel = channel1
agent.sinks.sink2.channel = channel2

My error log says

14/03/18 13:32:10 ERROR file.FileChannel: Failed to start the file channel [channel=channel1]

java.io.IOException：无法锁定 /home/.flume/file-channel/checkpoint。该目录已被锁定。 [频道=频道1] 在 org.apache.flume.channel.file.Log.lock(Log.java:802) 在 org.apache.flume.channel.file.Log.(Log.java:201) 在 org.apache.flume.channel.file.Log.(Log.java:66) 在 org.apache.flume.channel.file.Log$Builder.build(Log.java:167) 在 org.apache.flume.channel.file.FileChannel.start(FileChannel.java:242) 在 org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:236) 在 java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441) 在 java.util.concurrent.FutureTask$Sync.innerRunAndReset(FutureTask.java:317) 在 java.util.concurrent.FutureTask.runAndReset(FutureTask.java:150) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$101(ScheduledThreadPoolExecutor.java:98) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.runPeriodic(ScheduledThreadPoolExecutor.java:180) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:204) 在 java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908) 在 java.lang.Thread.run(Thread.java:662)

Answer 1

你不能在同一个目录中使用两个文件通道，因为文件通道目录中有一个写锁。你需要为每个文件通道指定不同的目录。例如：

a1.channels.c1.checkpointDir = /mnt/flume/checkpoint
a1.channels.c1.dataDirs = /mnt/flume/data
a1.channels.c2.checkpointDir = /mnt/flume2/checkpoint
a1.channels.c2.dataDirs = /mnt/flume2/data