I actually figured out the reason.. My thread class was defined as a static class, and I passed a
ChannelProcessor channel = getChannelProcessor(); to its constructor. Not sure why this does not work. After I remove the static for the thread class, and put ChannelProcessor channel = getChannelProcessor(); as a field inside the class, it starts working... On Wed, Jan 8, 2014 at 5:38 PM, Chen Wang <[email protected]>wrote: > David, > I am using flume-ng 1.4.0. > Thanks, > Chen > > > On Wed, Jan 8, 2014 at 5:26 PM, David Sinclair < > [email protected]> wrote: > >> What version of flume are you using? >> >> >> On Wed, Jan 8, 2014 at 8:01 PM, Chen Wang <[email protected]>wrote: >> >>> Hey, Guys, >>> First time using flume. It goes all well, however, I found that in my >>> custom source when I am calling >>> >>> Event event = EventBuilder.withBody(mySignal.toString().getBytes(), >>> headers); >>> >>> try { >>> >>> logger.debug("submitting to channel"); >>> >>> channel.processEvent(event); >>> >>> }catch (ChannelException e) { >>> >>> logger.debug("failed to submit to channel, exception {}",e); >>> >>> throw e; >>> >>> } >>> >>> >>> It just hangs there on "submitting to channel", and hdfs file apparently >>> is not created. Just wondering how can I trouble shoot this? I am running >>> it as >>> >>> bin/flume-ng agent -c conf -f conf/flume.conf -n SocketAgent >>> -Dflume.root.logger=DEBUG,console >>> >>> and my config file looks like: >>> >>> SocketAgent.sinks.HDFS.channel = MemChannel >>> >>> SocketAgent.sinks.HDFS.type = hdfs >>> >>> SocketAgent.sinks.HDFS.hdfs.path = >>> maprfs:///hive/myname.db/targeting_storm/%{epoch} >>> >>> SocketAgent.sinks.HDFS.hdfs.fileType = SequenceFile >>> >>> SocketAgent.sinks.HDFS.hdfs.writeFormat = Text >>> >>> SocketAgent.sinks.HDFS.hdfs.batchSize = 1000 >>> >>> SocketAgent.sinks.HDFS.hdfs.rollSize = 0 >>> >>> SocketAgent.sinks.HDFS.hdfs.rollCount = 10000 >>> >>> SocketAgent.sinks.HDFS.hdfs.rollInterval = 600 >>> >>> >>> SocketAgent.channels.MemChannel.type = memory >>> >>> SocketAgent.channels.MemChannel.capacity = 10000 >>> >>> SocketAgent.channels.MemChannel.transactionCapacity = 100 >>> >>> >>> Thank you very much!! >>> >>> Chen >>> >> >> >
