Search code examples
javahadoopflumeavro

Flume: org.apache.avro.ipc.NettyServer: Unexpected exception from downstream. java.nio.channels.ClosedChannelException

How to solve the problem? When I config the flume server, it had the below questiosns.

2014-10-20 22:24:01,480 INFO org.apache.avro.ipc.NettyServer: [id: 0x2fe09f1a, /ip:57063 => /ip:34001] OPEN
2014-10-20 22:24:01,481 INFO org.apache.avro.ipc.NettyServer: [id: 0x2fe09f1a, /ip:57063 => /ip:34001] BOUND: /ip:34001
2014-10-20 22:24:01,481 INFO org.apache.avro.ipc.NettyServer: [id: 0x2fe09f1a, /10.182.4.70:57063 => /ip:34001] CONNECTED: /ip:57063
2014-10-20 22:24:01,481 INFO org.apache.avro.ipc.NettyServer: [id: 0x2fe09f1a, /ip:57063 :> /ip:34001] DISCONNECTED
2014-10-20 22:24:01,481 INFO org.apache.avro.ipc.NettyServer: [id: 0x2fe09f1a, /ip:57063 :> /10.182.4.79:34001] UNBOUND
2014-10-20 22:24:01,481 INFO org.apache.avro.ipc.NettyServer: [id: 0x2fe09f1a, /10.182.4.70:57063 :> /10.182.4.79:34001] CLOSED
2014-10-20 22:24:01,481 INFO org.apache.avro.ipc.NettyServer: Connection to /10.182.4.70:57063 disconnected.
2014-10-20 22:24:01,481 WARN org.apache.avro.ipc.NettyServer: Unexpected exception from downstream.
java.nio.channels.ClosedChannelException
        at org.jboss.netty.channel.socket.nio.AbstractNioWorker.cleanUpWriteBuffer(AbstractNioWorker.java:673)
        at org.jboss.netty.channel.socket.nio.AbstractNioWorker.writeFromUserCode(AbstractNioWorker.java:400)
        at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.handleAcceptedSocket(NioServerSocketPipelineSink.java:120)
        at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.eventSunk(NioServerSocketPipelineSink.java:59)
        at org.jboss.netty.channel.Channels.write(Channels.java:733)
        at org.jboss.netty.channel.Channels.write(Channels.java:694)
        at org.jboss.netty.handler.codec.compression.ZlibEncoder.finishEncode(ZlibEncoder.java:380)
        at org.jboss.netty.handler.codec.compression.ZlibEncoder.handleDownstream(ZlibEncoder.java:316)
        at org.jboss.netty.handler.codec.oneone.OneToOneEncoder.handleDownstream(OneToOneEncoder.java:55)
        at org.jboss.netty.channel.Channels.close(Channels.java:821)

And the flume.conf is as belows. 

instance_35001.channels.channel1.checkpointDir=editlog/checkpoint
instance_35001.channels.channel1.dataDirs=editlog/data
instance_35001.channels.channel1.capacity=200000000
instance_35001.channels.channel1.transactionCapacity=1000000
instance_35001.channels.channel1.checkpointInterval=10000

instance_35001.sources=source1
instance_35001.sources.source1.type=avro
instance_35001.sources.source1.bind=0.0.0.0
instance_35001.sources.source1.port=34001
instance_35001.sources.source1.compression-type=deflate
instance_35001.sources.source1.channels=channel1

instance_35001.sources.source1.interceptors = inter1
instance_35001.sources.source1.interceptors.inter1.type = host
instance_35001.sources.source1.interceptors.inter1.hostHeader = servername

instance_35001.sinks=sink1

instance_35001.sinks.sink1.type=hdfs
instance_35001.sinks.sink1.hdfs.path=hdfs://address:5000/user/admin/%{appname}/%Y/%m/%d/
instance_35001.sinks.sink1.hdfs.filePrefix=%{appname}-%{hostname}-%{servername}.34001
instance_35001.sinks.sink1.hdfs.rollInterval=0
instance_35001.sinks.sink1.hdfs.rollCount=0
instance_35001.sinks.sink1.hdfs.rollSize=21521880492

The environment is CDH5. And the sink is the hdfs program. The log is usually very normal. but the sink is very slowly. So please help me. Thanks.

Solution

  • One thing I can see here is that your roll size is significantly larger than the channel capacity. So before rolling the file everything is stored in the channel which gets filled up after a point and starts throwing an error.

    instance_35001.channels.channel1.capacity=200000000

instance_35001.sinks.sink1.hdfs.rollSize=21521880492

    Keep your role size around the block size you have set for HDFS. Also HDFS sink has a default batch size of 100. Change it to some larger value and see how it behaves.