Hi

I am trying to setup a configuration where a single source agent load
balances between two aggregate agents which multiplex the flow to two end
points. I don't think I have the channel capacity properly configured but
flume always seems to end up hanging for me. If all channels are at
capacity should the source try and send again once the they have emptied?

Here is my configuration.

#
# Properties of memoryChannel
#
local_agent.channels.memoryChannel-1.type = memory
local_agent.channels.memoryChannel-1.capacity = 100000
local_agent.channels.memoryChannel-1.transactionCapacity = 1000

collector_agent_1.channels.memoryChannel-1.type = memory
collector_agent_1.channels.memoryChannel-1.capacity = 100000
collector_agent_1.channels.memoryChannel-1.transactionCapacity = 1000

collector_agent_1.channels.memoryChannel-2.type = memory
collector_agent_1.channels.memoryChannel-2.capacity = 100000
collector_agent_1.channels.memoryChannel-2.transactionCapacity = 1000

collector_agent_2.channels.memoryChannel-1.type = memory
collector_agent_2.channels.memoryChannel-1.capacity = 100000
collector_agent_2.channels.memoryChannel-1.transactionCapacity = 1000

collector_agent_2.channels.memoryChannel-2.type = memory
collector_agent_2.channels.memoryChannel-2.capacity = 100000
collector_agent_2.channels.memoryChannel-2.transactionCapacity = 1000

#
# Properties for spooling directory source
#
local_agent.sources.spooldir-1.type = spooldir
local_agent.sources.spooldir-1.spoolDir = ~/flume_test/ready
local_agent.sources.spooldir-1.fileHeader = true
local_agent.sources.spooldir-1.channels = memoryChannel-1

#
# Properties for the avro sink 1 agent to collector 1
#
local_agent.sinks.avroSink-1.type = avro
local_agent.sinks.avroSink-1.hostname = 127.0.0.1
local_agent.sinks.avroSink-1.port = 4545
local_agent.sinks.avroSink-1.channel = memoryChannel-1

#
# Properties for the avro sink agent to collector 2
#
local_agent.sinks.avroSink-2.type = avro
local_agent.sinks.avroSink-2.hostname = 127.0.0.1
local_agent.sinks.avroSink-2.port = 4546
local_agent.sinks.avroSink-2.channel = memoryChannel-1

#
# Properties for the avro source collector 1
#
collector_agent_1.sources.avroSource-1.type = avro
collector_agent_1.sources.avroSource-1.bind = 127.0.0.1
collector_agent_1.sources.avroSource-1.port = 4545
collector_agent_1.sources.avroSource-1.channels = memoryChannel-1
memoryChannel-2

#
# Properties for the avro source collector 2
#
collector_agent_2.sources.avroSource-2.type = avro
collector_agent_2.sources.avroSource-2.bind = 127.0.0.1
collector_agent_2.sources.avroSource-2.port = 4546
collector_agent_2.sources.avroSource-2.channels = memoryChannel-1
memoryChannel-2

# End points for collector 1

# ElasticSearch endpoint collector 1

collector_agent_1.sinks.elastic-search-sink-1.type org.apache.flume.sink.elasticsearch.ElasticSearchSink
collector_agent_1.sinks.elastic-search-sink-1.hostNames = 127.0.0.1:9300
collector_agent_1.sinks.elastic-search-sink-1.clusterName = elasticsearch
collector_agent_1.sinks.elastic-search-sink-1.batchSize = 10
collector_agent_1.sinks.elastic-search-sink-1.channel = memoryChannel-1

# HDFS endpoint collector 1

collector_agent_1.sinks.sink1.type = hdfs
collector_agent_1.sinks.sink1.hdfs.path = hdfs://
hadoop-name-node1.dc1.ci-mang.van.dev.net:8020/flumeng_test
collector_agent_1.sinks.sink1.hdfs.fileType = DataStream
collector_agent_1.sinks.sink1.hdfs.rollInterval = 300
collector_agent_1.sinks.sink1.hdfs.rollSize = 0
collector_agent_1.sinks.sink1.hdfs.rollCount = 0
collector_agent_1.sinks.sink1.hdfs.batchSize = 1000
collector_agent_1.sinks.sink1.txnEventMax = 1000
collector_agent_1.sinks.sink1.serializer = avro_event
collector_agent_1.sinks.sink1.channel = memoryChannel-2

# ElasticSearch endpoint collector 2

collector_agent_2.sinks.elastic-search-sink-1.type org.apache.flume.sink.elasticsearch.ElasticSearchSink
collector_agent_2.sinks.elastic-search-sink-1.hostNames = 127.0.0.1:9300
collector_agent_2.sinks.elastic-search-sink-1.clusterName = elasticsearch
collector_agent_2.sinks.elastic-search-sink-1.batchSize = 10
collector_agent_2.sinks.elastic-search-sink-1.channel = memoryChannel-1

# HDFS endpoint collector 2

collector_agent_2.sinks.sink1.type = hdfs
collector_agent_2.sinks.sink1.hdfs.path = hdfs://
hadoop-name-node1.dc1.ci-mang.van.dev.net:8020/flumeng_test_3
collector_agent_2.sinks.sink1.hdfs.fileType = DataStream
collector_agent_2.sinks.sink1.hdfs.rollInterval = 300
collector_agent_2.sinks.sink1.hdfs.rollSize = 0
collector_agent_2.sinks.sink1.hdfs.rollCount = 0
collector_agent_2.sinks.sink1.hdfs.batchSize = 1000
collector_agent_2.sinks.sink1.txnEventMax = 1000
collector_agent_2.sinks.sink1.serializer = avro_event
collector_agent_2.sinks.sink1.channel = memoryChannel-2

# Specify priorities for the sinks on the agent

local_agent.sinkgroups.ha.sinks = avroSink-1 avroSink-2
local_agent.sinkgroups.ha.processor.type = failover
local_agent.sinkgroups.ha.priority.avroSink-1 = 2
local_agent.sinkgroups.ha.priority.avroSink-2 = 1

# Wire the source agents up

local_agent.sources = spooldir-1
local_agent.sinks = avroSink-1 avroSink-2
local_agent.sinkgroups = ha
local_agent.channels = memoryChannel-1

# Wire the collector agents up

collector_agent_1.sources = avroSource-1
collector_agent_1.sinks = elastic-search-sink-1 sink1
collector_agent_1.channels = memoryChannel-1 memoryChannel-2

collector_agent_2.sources = avroSource-2
collector_agent_2.sinks = elastic-search-sink-1 sink1
collector_agent_2.channels = memoryChannel-1 memoryChannel-2

I will get the following exceptions on the collector nodes

*org.apache.flume.ChannelException: Unable to put batch on required
channel: org.apache.flume.channel.MemoryChannel{name: memoryChannel-1}*
* at
org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:200)
*
* at org.apache.flume.source.AvroSource.appendBatch(AvroSource.java:259)*
* at sun.reflect.GeneratedMethodAccessor6.invoke(Unknown Source)*
* at
sun.reflect.Deleg

Hi,

Can you confirm that ElasticSearchSink is consuming events?

> Once this happens flume never resumes.

It doesn't bring anything after that? Even minutes later it's still
doing nothing?

brock

On Tue, Oct 16, 2012 at 1:23 PM, Cameron Gandevia <[EMAIL PROTECTED]> wrote:
> Hi
>
> I am trying to setup a configuration where a single source agent load
> balances between two aggregate agents which multiplex the flow to two end
> points. I don't think I have the channel capacity properly configured but
> flume always seems to end up hanging for me. If all channels are at capacity
> should the source try and send again once the they have emptied?
>
> Here is my configuration.
>
> #
> # Properties of memoryChannel
> #
> local_agent.channels.memoryChannel-1.type = memory
> local_agent.channels.memoryChannel-1.capacity = 100000
> local_agent.channels.memoryChannel-1.transactionCapacity = 1000
>
> collector_agent_1.channels.memoryChannel-1.type = memory
> collector_agent_1.channels.memoryChannel-1.capacity = 100000
> collector_agent_1.channels.memoryChannel-1.transactionCapacity = 1000
>
> collector_agent_1.channels.memoryChannel-2.type = memory
> collector_agent_1.channels.memoryChannel-2.capacity = 100000
> collector_agent_1.channels.memoryChannel-2.transactionCapacity = 1000
>
> collector_agent_2.channels.memoryChannel-1.type = memory
> collector_agent_2.channels.memoryChannel-1.capacity = 100000
> collector_agent_2.channels.memoryChannel-1.transactionCapacity = 1000
>
> collector_agent_2.channels.memoryChannel-2.type = memory
> collector_agent_2.channels.memoryChannel-2.capacity = 100000
> collector_agent_2.channels.memoryChannel-2.transactionCapacity = 1000
>
> #
> # Properties for spooling directory source
> #
> local_agent.sources.spooldir-1.type = spooldir
> local_agent.sources.spooldir-1.spoolDir = ~/flume_test/ready
> local_agent.sources.spooldir-1.fileHeader = true
> local_agent.sources.spooldir-1.channels = memoryChannel-1
>
> #
> # Properties for the avro sink 1 agent to collector 1
> #
> local_agent.sinks.avroSink-1.type = avro
> local_agent.sinks.avroSink-1.hostname = 127.0.0.1
> local_agent.sinks.avroSink-1.port = 4545
> local_agent.sinks.avroSink-1.channel = memoryChannel-1
>
> #
> # Properties for the avro sink agent to collector 2
> #
> local_agent.sinks.avroSink-2.type = avro
> local_agent.sinks.avroSink-2.hostname = 127.0.0.1
> local_agent.sinks.avroSink-2.port = 4546
> local_agent.sinks.avroSink-2.channel = memoryChannel-1
>
> #
> # Properties for the avro source collector 1
> #
> collector_agent_1.sources.avroSource-1.type = avro
> collector_agent_1.sources.avroSource-1.bind = 127.0.0.1
> collector_agent_1.sources.avroSource-1.port = 4545
> collector_agent_1.sources.avroSource-1.channels = memoryChannel-1
> memoryChannel-2
>
> #
> # Properties for the avro source collector 2
> #
> collector_agent_2.sources.avroSource-2.type = avro
> collector_agent_2.sources.avroSource-2.bind = 127.0.0.1
> collector_agent_2.sources.avroSource-2.port = 4546
> collector_agent_2.sources.avroSource-2.channels = memoryChannel-1
> memoryChannel-2
>
> # End points for collector 1
>
> # ElasticSearch endpoint collector 1
>
> collector_agent_1.sinks.elastic-search-sink-1.type > org.apache.flume.sink.elasticsearch.ElasticSearchSink
> collector_agent_1.sinks.elastic-search-sink-1.hostNames = 127.0.0.1:9300
> collector_agent_1.sinks.elastic-search-sink-1.clusterName = elasticsearch
> collector_agent_1.sinks.elastic-search-sink-1.batchSize = 10
> collector_agent_1.sinks.elastic-search-sink-1.channel = memoryChannel-1
>
> # HDFS endpoint collector 1
>
> collector_agent_1.sinks.sink1.type = hdfs
> collector_agent_1.sinks.sink1.hdfs.path > hdfs://hadoop-name-node1.dc1.ci-mang.van.dev.net:8020/flumeng_test
> collector_agent_1.sinks.sink1.hdfs.fileType = DataStream
> collector_agent_1.sinks.sink1.hdfs.rollInterval = 300
> collector_agent_1.sinks.sink1.hdfs.rollSize = 0
> collector_agent_1.sinks.sink1.hdfs.rollCount = 0
> collector_agent_1.sinks.sink1.hdfs.batchSize = 1000

Apache MRUnit - Unit testing MapReduce - http://incubator.apache.org/mrunit/

I changed my configuration to write to two HDFS instances to eliminate the
ElasticSearch sink.

I noticed the following exception on the source agent.

*2012-10-16 12:03:49,989 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[DEBUG -
org.apache.flume.sink.AvroSink.destroyConnection(AvroSink.java:199)] Avro
sink avroSink-1 closing avro client: NettyAvroRpcClient { host: localhost,
port: 4545 }*
*2012-10-16 12:03:49,993 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[WARN -
org.apache.flume.sink.FailoverSinkProcessor.process(FailoverSinkProcessor.java:185)]
Sink avroSink-1 failed and has been sent to failover list*
*org.apache.flume.EventDeliveryException: Failed to send events*
* at org.apache.flume.sink.AvroSink.process(AvroSink.java:325)*
* at
org.apache.flume.sink.FailoverSinkProcessor.process(FailoverSinkProcessor.java:182)
*
* at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)*
* at java.lang.Thread.run(Thread.java:662)*
*Caused by: org.apache.flume.EventDeliveryException: NettyAvroRpcClient {
host: localhost, port: 4545 }: Failed to send batch*
* at
org.apache.flume.api.NettyAvroRpcClient.appendBatch(NettyAvroRpcClient.java:236)
*
* at org.apache.flume.sink.AvroSink.process(AvroSink.java:309)*
* ... 3 more*
*Caused by: org.apache.flume.EventDeliveryException: NettyAvroRpcClient {
host: localhost, port: 4545 }: Avro RPC call returned Status: FAILED*
* at
org.apache.flume.api.NettyAvroRpcClient.waitForStatusOK(NettyAvroRpcClient.java:312)
*
* at
org.apache.flume.api.NettyAvroRpcClient.appendBatch(NettyAvroRpcClient.java:295)
*
* at
org.apache.flume.api.NettyAvroRpcClient.appendBatch(NettyAvroRpcClient.java:224)
*
* ... 4 more*
*2012-10-16 12:03:51,998 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[INFO - org.apache.flume.sink.AvroSink.createConnection(AvroSink.java:178)]
Avro sink avroSink-1: Building RpcClient with hostname: 127.0.0.1, port:
4545*
*2012-10-16 12:03:51,999 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[DEBUG -
org.apache.flume.api.NettyAvroRpcClient.configure(NettyAvroRpcClient.java:420)]
Batch size string = null*
*2012-10-16 12:03:52,001 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[DEBUG -
org.apache.flume.sink.AvroSink.createConnection(AvroSink.java:192)] Avro
sink avroSink-1: Created RpcClient: NettyAvroRpcClient { host: localhost,
port: 4545 }*
*2012-10-16 12:03:55,008 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[DEBUG -
org.apache.flume.sink.AvroSink.destroyConnection(AvroSink.java:199)] Avro
sink avroSink-1 closing avro client: NettyAvroRpcClient { host: localhost,
port: 4545 }*
*2012-10-16 12:03:55,009 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[DEBUG -
org.apache.flume.sink.FailoverSinkProcessor$FailedSink.incFails(FailoverSinkProcessor.java:97)]
Sink avroSink-1 failed again, new refresh is at 1350414239009, current time
1350414235009*
*2012-10-16 12:03:59,341 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[INFO - org.apache.flume.sink.AvroSink.createConnection(AvroSink.java:178)]
Avro sink avroSink-1: Building RpcClient with hostname: 127.0.0.1, port:
4545*
*2012-10-16 12:03:59,342 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[DEBUG -
org.apache.flume.api.NettyAvroRpcClient.configure(NettyAvroRpcClient.java:420)]
Batch size string = null*
*2012-10-16 12:03:59,347 (SinkRunner-PollingRunner-FailoverSinkProcessor)
[DEBUG -
org.apache.flume.sink.AvroSink.createConnection(AvroSink.java:192)] Avro
sink avroSink-1: Created RpcClient: NettyAvroRpcClient { host: localhost,
port: 4545 }*
*2012-10-16 12:04:13,849 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.file.AbstractFileConfigurationProvider$FileWatcherRunnable.run(AbstractFileConfigurationProvider.java:188)]
Checking file:../conf/flume.conf for changes*
*2012-10-16 12:04:43,850 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.file.AbstractFileConfigurationProvider$FileWatcherRunnable.run(AbstractFileConfigurationProvider.java:188)]
Checking file:../conf/flume.conf for changes*
*2012-10-16 12:05:13,850 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.file.AbstractFileConfigurationProvider$FileWatcherRunnable.run(AbstractFileConfigurationProvider.java:188)]
Checking file:../conf/flume.conf for changes*

Then on the first collector

*2012-10-16 12:03:52,003 (New I/O server worker #1-2) [DEBUG -
org.apache.flume.source.AvroSource.appendBatch(AvroSource.java:244)] Avro
source avroSource-1: Received avro event batch of 100 events.*
*2012-10-16 12:03:55,004 (New I/O server worker #1-2) [ERROR -
org.apache.flume.source.AvroSource.appendBatch(AvroSource.java:261)] Avro
source avroSource-1: Unable to process event batch. Exception follows.*
*org.apache.flume.ChannelException: Unable to put batch on required
channel: org.apache.flume.channel.MemoryChannel{name: memoryChannel-1}*
* at
org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:200)
*
* at org.apache.flume.source.AvroSource.appendBatch(AvroSource.java:259)*
* at sun.reflect.GeneratedMethodAccessor6.invoke(Unknown Source)*
* at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
*
* at java.lang.reflect.Method.invoke(Method.java:597)*
* at
org.apache.avro.ipc.specific.SpecificResponder.respond(SpecificResponder.java:88)
*
* at org.apache.avro.ipc.Responder.respond(Responder.java:149)*
* at
org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.messageReceived(NettyServer.java:188)
*
* at
org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:80)
*
* at
org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:173)
*
* at
org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
*
* at
org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:783)
*
* at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:302)
*
* at
org.jboss.netty.handler.codec.frame.FrameDecoder

With the configuration below, I was able to kill collector1, see
collector2 take over and reverse.

Note that I had to decrease the capacity of the channels significantly
because I was using a smaller heap size. Before decreasing the
capacity, a few threads in the local_agent ran of memory and then
behaved like you explained.

Brock

#
# Properties of memoryChannel
#
local_agent.channels.memoryChannel-1.type = memory
local_agent.channels.memoryChannel-1.capacity = 100000
local_agent.channels.memoryChannel-1.transactionCapacity = 1000

collector_agent_1.channels.memoryChannel-1.type = memory
collector_agent_1.channels.memoryChannel-1.capacity = 100000
collector_agent_1.channels.memoryChannel-1.transactionCapacity = 1000

collector_agent_1.channels.memoryChannel-2.type = memory
collector_agent_1.channels.memoryChannel-2.capacity = 100000
collector_agent_1.channels.memoryChannel-2.transactionCapacity = 1000

collector_agent_2.channels.memoryChannel-1.type = memory
collector_agent_2.channels.memoryChannel-1.capacity = 100000
collector_agent_2.channels.memoryChannel-1.transactionCapacity = 1000

collector_agent_2.channels.memoryChannel-2.type = memory
collector_agent_2.channels.memoryChannel-2.capacity = 100000
collector_agent_2.channels.memoryChannel-2.transactionCapacity = 1000

#
# Properties for spooling directory source
#
local_agent.sources.spooldir-1.type = seq
local_agent.sources.spooldir-1.channels = memoryChannel-1

#
# Properties for the avro sink 1 agent to collector 1
#
local_agent.sinks.avroSink-1.type = avro
local_agent.sinks.avroSink-1.hostname = 127.0.0.1
local_agent.sinks.avroSink-1.port = 4545
local_agent.sinks.avroSink-1.channel = memoryChannel-1

#
# Properties for the avro sink agent to collector 2
#
local_agent.sinks.avroSink-2.type = avro
local_agent.sinks.avroSink-2.hostname = 127.0.0.1
local_agent.sinks.avroSink-2.port = 4546
local_agent.sinks.avroSink-2.channel = memoryChannel-1

#
# Properties for the avro source collector 1
#
collector_agent_1.sources.avroSource-1.type = avro
collector_agent_1.sources.avroSource-1.bind = 127.0.0.1
collector_agent_1.sources.avroSource-1.port = 4545
collector_agent_1.sources.avroSource-1.channels = memoryChannel-1
memoryChannel-2

#
# Properties for the avro source collector 2
#
collector_agent_2.sources.avroSource-2.type = avro
collector_agent_2.sources.avroSource-2.bind = 127.0.0.1
collector_agent_2.sources.avroSource-2.port = 4546
collector_agent_2.sources.avroSource-2.channels = memoryChannel-1
memoryChannel-2

# End points for collector 1

# ElasticSearch endpoint collector 1

collector_agent_1.sinks.elastic-search-sink-1.type = null
collector_agent_1.sinks.elastic-search-sink-1.channel = memoryChannel-1

# HDFS endpoint collector 1

collector_agent_1.sinks.sink1.type = null
collector_agent_1.sinks.sink1.channel = memoryChannel-2

# ElasticSearch endpoint collector 2

collector_agent_2.sinks.elastic-search-sink-1.type = null
collector_agent_2.sinks.elastic-search-sink-1.channel = memoryChannel-1

# HDFS endpoint collector 2

collector_agent_2.sinks.sink1.type = null
collector_agent_2.sinks.sink1.channel = memoryChannel-2

# Specify priorities for the sinks on the agent

local_agent.sinkgroups.ha.sinks = avroSink-1 avroSink-2
local_agent.sinkgroups.ha.processor.type = failover
local_agent.sinkgroups.ha.priority.avroSink-1 = 2
local_agent.sinkgroups.ha.priority.avroSink-2 = 1

# Wire the source agents up

local_agent.sources = spooldir-1
local_agent.sinks = avroSink-1 avroSink-2
local_agent.sinkgroups = ha
local_agent.channels = memoryChannel-1

# Wire the collector agents up

collector_agent_1.sources = avroSource-1
collector_agent_1.sinks = elastic-search-sink-1 sink1
collector_agent_1.channels = memoryChannel-1 memoryChannel-2

collector_agent_2.sources = avroSource-2
collector_agent_2.sinks = elastic-search-sink-1 sink1
collector_agent_2.channels = memoryChannel-1 memoryChannel-2

Hey

Thanks for the reply. I think the problem is in the Spooling Directory
Source. I am not sure it recovers properly, once the failure happens I
don't see its thread pool running in the dump. I have added a question to
the code review which may be related but I don't understand how the sources
are restarted during failures to know if its the problem for sure. I will
keep looking and post anything I find.

Thanks

On Tue, Oct 16, 2012 at 12:45 PM, Brock Noland <[EMAIL PROTECTED]> wrote:

> With the configuration below, I was able to kill collector1, see
> collector2 take over and reverse.
>
> Note that I had to decrease the capacity of the channels significantly
> because I was using a smaller heap size. Before decreasing the
> capacity, a few threads in the local_agent ran of memory and then
> behaved like you explained.
>
> Brock
>
> #
> # Properties of memoryChannel
> #
> local_agent.channels.memoryChannel-1.type = memory
> local_agent.channels.memoryChannel-1.capacity = 100000
> local_agent.channels.memoryChannel-1.transactionCapacity = 1000
>
> collector_agent_1.channels.memoryChannel-1.type = memory
> collector_agent_1.channels.memoryChannel-1.capacity = 100000
> collector_agent_1.channels.memoryChannel-1.transactionCapacity = 1000
>
> collector_agent_1.channels.memoryChannel-2.type = memory
> collector_agent_1.channels.memoryChannel-2.capacity = 100000
> collector_agent_1.channels.memoryChannel-2.transactionCapacity = 1000
>
> collector_agent_2.channels.memoryChannel-1.type = memory
> collector_agent_2.channels.memoryChannel-1.capacity = 100000
> collector_agent_2.channels.memoryChannel-1.transactionCapacity = 1000
>
> collector_agent_2.channels.memoryChannel-2.type = memory
> collector_agent_2.channels.memoryChannel-2.capacity = 100000
> collector_agent_2.channels.memoryChannel-2.transactionCapacity = 1000
>
> #
> # Properties for spooling directory source
> #
> local_agent.sources.spooldir-1.type = seq
> local_agent.sources.spooldir-1.channels = memoryChannel-1
>
> #
> # Properties for the avro sink 1 agent to collector 1
> #
> local_agent.sinks.avroSink-1.type = avro
> local_agent.sinks.avroSink-1.hostname = 127.0.0.1
> local_agent.sinks.avroSink-1.port = 4545
> local_agent.sinks.avroSink-1.channel = memoryChannel-1
>
> #
> # Properties for the avro sink agent to collector 2
> #
> local_agent.sinks.avroSink-2.type = avro
> local_agent.sinks.avroSink-2.hostname = 127.0.0.1
> local_agent.sinks.avroSink-2.port = 4546
> local_agent.sinks.avroSink-2.channel = memoryChannel-1
>
> #
> # Properties for the avro source collector 1
> #
> collector_agent_1.sources.avroSource-1.type = avro
> collector_agent_1.sources.avroSource-1.bind = 127.0.0.1
> collector_agent_1.sources.avroSource-1.port = 4545
> collector_agent_1.sources.avroSource-1.channels = memoryChannel-1
> memoryChannel-2
>
> #
> # Properties for the avro source collector 2
> #
> collector_agent_2.sources.avroSource-2.type = avro
> collector_agent_2.sources.avroSource-2.bind = 127.0.0.1
> collector_agent_2.sources.avroSource-2.port = 4546
> collector_agent_2.sources.avroSource-2.channels = memoryChannel-1
> memoryChannel-2
>
> # End points for collector 1
>
> # ElasticSearch endpoint collector 1
>
> collector_agent_1.sinks.elastic-search-sink-1.type = null
> collector_agent_1.sinks.elastic-search-sink-1.channel = memoryChannel-1
>
> # HDFS endpoint collector 1
>
> collector_agent_1.sinks.sink1.type = null
> collector_agent_1.sinks.sink1.channel = memoryChannel-2
>
> # ElasticSearch endpoint collector 2
>
> collector_agent_2.sinks.elastic-search-sink-1.type = null
> collector_agent_2.sinks.elastic-search-sink-1.channel = memoryChannel-1
>
> # HDFS endpoint collector 2
>
> collector_agent_2.sinks.sink1.type = null
> collector_agent_2.sinks.sink1.channel = memoryChannel-2
>
> # Specify priorities for the sinks on the agent
>
> local_agent.sinkgroups.ha.sinks = avroSink-1 avroSink-2
> local_agent.sinkgroups.ha.processor.type = failover
> local_agent.sinkgroups.ha.priority.avroSink-1 = 2

Thanks

Cameron Gandevia

Hi Cameron, I had a look at your logs and what would be happening:

- Looks like your first collector either isn't sinking events, or it's
just too slow at it.
- By the look of the logs the second one *is* delivering events. The
failover processor doesn't actually say anything if nothing is wrong.
Perhaps adding some logging to
FailoverSinkProcessor.moveActiveToDeadAndGetNext() would make this more
user friendly. Could you check if stuff was actually delivered by
collector 2?
- The json output for avro-sink2 looks like it is successfully sending
data. What's the output like for the second collector?
- I'd recommend setting up ganglia to verify the throughput on each
component to see if there's an imbalance. Those JSON stats quite frankly
look a bit odd. It doesn't look like the channels are full, which would
indicate that the batch size is greater than the transaction capacity.
But this doesn't appear to be the case in the configuration, unless we
have a bug expecting batchsize < transactionCap rather than <=. It seems
some batch sizes aren't specified(so using default. I'm pretty sure none
of the defaults are > 1000 but you might want to check that just to be safe

So if this isn't resolved: Check to see if collector 2 is receiving
events, if not, try chancing transaction size to be greater than the
batch size. If there's still a problem, thread dumps from all three
agents would help a lot, and if you can, have a look at the stats from
ganglia.

On 10/17/2012 04:53 AM, Cameron Gandevia wrote:
> Hey
>
> Thanks for the reply. I think the problem is in the Spooling Directory
> Source. I am not sure it recovers properly, once the failure happens I
> don't see its thread pool running in the dump. I have added a question
> to the code review which may be related but I don't understand how the
> sources are restarted during failures to know if its the problem for
> sure. I will keep looking and post anything I find.
>
> Thanks
>
> On Tue, Oct 16, 2012 at 12:45 PM, Brock Noland <[EMAIL PROTECTED]
> <mailto:[EMAIL PROTECTED]>> wrote:
>
>     With the configuration below, I was able to kill collector1, see
>     collector2 take over and reverse.
>
>     Note that I had to decrease the capacity of the channels significantly
>     because I was using a smaller heap size. Before decreasing the
>     capacity, a few threads in the local_agent ran of memory and then
>     behaved like you explained.
>
>     Brock
>
>     #
>     # Properties of memoryChannel
>     #
>     local_agent.channels.memoryChannel-1.type = memory
>     local_agent.channels.memoryChannel-1.capacity = 100000
>     local_agent.channels.memoryChannel-1.transactionCapacity = 1000
>
>     collector_agent_1.channels.memoryChannel-1.type = memory
>     collector_agent_1.channels.memoryChannel-1.capacity = 100000
>     collector_agent_1.channels.memoryChannel-1.transactionCapacity = 1000
>
>     collector_agent_1.channels.memoryChannel-2.type = memory
>     collector_agent_1.channels.memoryChannel-2.capacity = 100000
>     collector_agent_1.channels.memoryChannel-2.transactionCapacity = 1000
>
>     collector_agent_2.channels.memoryChannel-1.type = memory
>     collector_agent_2.channels.memoryChannel-1.capacity = 100000
>     collector_agent_2.channels.memoryChannel-1.transactionCapacity = 1000
>
>     collector_agent_2.channels.memoryChannel-2.type = memory
>     collector_agent_2.channels.memoryChannel-2.capacity = 100000
>     collector_agent_2.channels.memoryChannel-2.transactionCapacity = 1000
>
>     #
>     # Properties for spooling directory source
>     #
>     local_agent.sources.spooldir-1.type = seq
>     local_agent.sources.spooldir-1.channels = memoryChannel-1
>
>     #
>     # Properties for the avro sink 1 agent to collector 1
>     #
>     local_agent.sinks.avroSink-1.type = avro
>     local_agent.sinks.avroSink-1.hostname = 127.0.0.1
>     local_agent.sinks.avroSink-1.port = 4545
>     local_agent.sinks.avroSink-1.channel = memoryChannel-1