I am trying out Flume, but I am getting an error:
2016-02-08 17:52:46,368 main DEBUG Shutdown hook enabled. Registering a new
one.
2016-02-08 17:52:46,370 main DEBUG LoggerContext[name=504bae78,
org.apache.logging.log4j.core.LoggerContext@5824a83d] started OK.
2016-02-08 17:52:46,388 main ERROR Recursive call to appender flume
2016-02-08 17:52:46,394 main ERROR Recursive call to appender flume
2016-02-08 17:52:46,464 main ERROR Recursive call to appender flume
2016-02-08 17:52:46,464 main ERROR Recursive call to appender flume
Flume output:
2016-02-08 17:46:51,334 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:372)]
Post validation configuration for agent
AgentConfiguration created without Configuration stubs for which only basic
syntactical validation was performed[agent]
SOURCES: {avroSrc={ parameters:{bind=localhost, channels=memoryChannel,
port=8800, type=avro} }}
CHANNELS: {memoryChannel={ parameters:{type=memory, capacity=100} }}
AgentConfiguration created with Configuration stubs for which full
validation was performed[agent]
SINKS: {loggerSink=ComponentConfiguration[loggerSink]
CONFIG:
CHANNEL:memoryChannel
}
2016-02-08 17:52:46,513 (New I/O worker #3) [INFO -
org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)]
[id: 0xe2275c51, /127.0.0.1:61033 => /127.0.0.1:8800] BOUND: /127.0.0.1:8800
2016-02-08 17:52:46,513 (New I/O worker #3) [INFO -
org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)]
[id: 0xe2275c51, /127.0.0.1:61033 => /127.0.0.1:8800] CONNECTED: /
127.0.0.1:61033
log4j2.xml:
<Flume name="flume" ignoreExceptions="false" mdcPrefix="ReqCtx_"
compress="false">
<Agent host="localhost" port="8800" />
<RFC5424Layout enterpriseNumber="12293" includeMDC="true"
mdcId="RequestContext" appName="GL" />
</Flume>
<Root level="debug">
<AppenderRef ref="flume"/>
</Root>
Flume config:
agent.sources = avroSrc
agent.channels = memoryChannel
agent.sinks = loggerSink
# For each one of the sources, the type is defined
# agent.sources.seqGenSrc.type = seq
agent.sources.avroSrc.type = avro
agent.sources.avroSrc.bind = localhost
agent.sources.avroSrc.port = 8800
# The channel can be defined as follows.
agent.sources.avroSrc.channels = memoryChannel
# Each sink's type must be defined
agent.sinks.loggerSink.type = logger
#Specify the channel the sink should use
agent.sinks.loggerSink.channel = memoryChannel
# Each channel's type is defined.
agent.channels.memoryChannel.type = memory
# Other config values specific to each type of channel(sink or source)
# can be defined as well
# In this case, it specifies the capacity of the memory channel
agent.channels.memoryChannel.capacity = 100
