Re: Mitigating uneven queue consumption

[John Heitmann <jh...@gmail.com>]

(moved to activemq-dev)

On Jun 19, 2006, at 11:41 PM, James Strachan wrote:

> The forced-reconnect after a certain amount of time is probably the
> simplest way of solving this issue.

I ended up implementing this first. It works ok, but it's pretty lame  
by design. It has actually turned out to be most handy to stress test  
the failover transport. Would you be interested in the patch for this  
or would you rather wait for multiplexing?

> I'm also interested in the multiplexing option. We already have the
> FanoutTransport which kinda does something like this; though we'd need
> to think carefully about the semantics, like transactions etc.

I am part way through the implementation of this and wanted to bounce  
the rough design off you and see if it sounds like the right  
direction. I'm calling it 'fanin', but I see you've called something  
similar 'jedi' in Jira. Maybe I'll rename mine 'sith' just to be  
contrarian :)

It currently connects to all discovered brokers. Message sends round- 
robin to individual brokers (ie only one broker handles a message  
send). Connection/session metadata commands are handled with the same  
state tracking failover uses. Anything stored in the state tracker is  
broadcast instead of unicast. Message acks are obviously only sent to  
the broker they originated from. Ranged acks are intelligently split  
so each broker gets the  acks for the bits of range only it knows about.

Transactions are still todo. My plan is that once the transport sees  
a transaction it punts and stops round-robin sending and locks down  
to one broker. That should be good enough for 1.0. Request/reply and  
temporary destinations are still untested, but I think they should  
work without much design tweaking.

I don't plan on doing anything about message ordering. We don't care  
much about ordering, and I don't think there's much that could be  
done here anyway.

Connection handling is done roughly the same way fanout does. The  
transport is basically a frankenstein of fanout and failover. Once I  
have the functionality complete I think it'll worthwhile to refactor  
the 3 a bit.

Does this sounds pretty reasonable?

John

> On 6/19/06, John Heitmann <jh...@gmail.com> wrote:
>> Our setup: We have a set of brokers directly feeding consumers
>> messages on a single queue destination (I'm going to lie a bit about
>> our setup to keep things simple). On each host we're running a broker
>> and multiple consumers. Each consumer connects to the broker set with
>> a failover connection. Messages flow into the broker set from
>> upstream sources (in our case another set of brokers) in a pretty
>> even fashion, so each broker gets about the same amount of messages.
>>
>> The problem with this is that it's easy for one broker to have an
>> uneven number of consumers, and in some cases have so few consumers
>> that it will backlog. One way this can happen is if a broker is taken
>> down for some routine reason. All the connections to that broker will
>> fail over to another broker. When the broker comes back up it will
>> not receive new connections until consumers restart yet it will
>> continue to receive an equal share of messages from upstream.
>>
>> We're considering implementing two different code changes to ActiveMQ
>> to help with this, but I wanted to poll the group to be sure there
>> was not already something there to help us.
>>
>> Our first solution would be to add logic to ActiveMQConnection that
>> would trigger a reconnect on the failover transport after a certain
>> configurable time period expires and the current transaction is
>> complete. This would give decent statistical certainty that a single
>> broker won't ever get a large backlog compared to the others and
>> should be transparent to the client code except for a hitch in
>> message flow.
>>
>> Our second solution is to implement connection multiplexing for
>> consumers. Behind the scenes each consumer would connect to all of
>> the broker set, divide its prefetch buffer across all those brokers,
>> and start grabbing messages from multiple at once. We'd also need
>> hooks to auto-discovery for when we bring new brokers up or old
>> brokers come back. Also transactions would probably need to be tied
>> to a single broker. This seems like it would be smoother behavior
>> overall, but tougher to implement than the first so we're putting it
>> off. Scalability is also a concern.
>>
>> If you know of a way to do this already I would appreciate hearing
>> about it. Thanks,
>>
>> John
>>
>
>
> -- 
>
> James
> -------
> http://radio.weblogs.com/0112098/

Re: Mitigating uneven queue consumption

[James Strachan <ja...@gmail.com>]

On 8/22/06, John Heitmann <jh...@gmail.com> wrote:
> (moved to activemq-dev)
>
> On Jun 19, 2006, at 11:41 PM, James Strachan wrote:
>
> > The forced-reconnect after a certain amount of time is probably the
> > simplest way of solving this issue.
>
> I ended up implementing this first. It works ok, but it's pretty lame
> by design. It has actually turned out to be most handy to stress test
> the failover transport. Would you be interested in the patch for this
> or would you rather wait for multiplexing?

Sure :)


> > I'm also interested in the multiplexing option. We already have the
> > FanoutTransport which kinda does something like this; though we'd need
> > to think carefully about the semantics, like transactions etc.
>
> I am part way through the implementation of this and wanted to bounce
> the rough design off you and see if it sounds like the right
> direction.

Great! :)

> I'm calling it 'fanin', but I see you've called something
> similar 'jedi' in Jira. Maybe I'll rename mine 'sith' just to be
> contrarian :)

LOL!


> It currently connects to all discovered brokers. Message sends round-
> robin to individual brokers (ie only one broker handles a message
> send).

BTW for topics you might wanna send the message to all brokers.


>  Connection/session metadata commands are handled with the same
> state tracking failover uses. Anything stored in the state tracker is
> broadcast instead of unicast. Message acks are obviously only sent to
> the broker they originated from. Ranged acks are intelligently split
> so each broker gets the  acks for the bits of range only it knows about.

Sounds *awesome*! Can't wait for the patch :)


> Transactions are still todo. My plan is that once the transport sees
> a transaction it punts and stops round-robin sending and locks down
> to one broker. That should be good enough for 1.0.

Yeah, that'd be fine to start with.

Another option is that each transaction is pinned to a specific broker
& all operations are fixed with that broker. Bear in mind different
threads can be performing different transactions - so if a Message /
MessageAck has a transactionId property then associate it with a
single broker. We could then round robin transactions around brokers
too :)


> Request/reply and
> temporary destinations are still untested, but I think they should
> work without much design tweaking.

Yeah - we'd need to broadcast temporary destination commands to all
brokers just in case (as the temporary destination could be put on the
JMSReplyTo on another message


> I don't plan on doing anything about message ordering. We don't care
> much about ordering, and I don't think there's much that could be
> done here anyway.

Cool. If forks are worried about that we could pin a destination to a
broker - or even a message for a destination with a specific Message
Group header (JMSXGroupID) to a broker so we could load balance
messages to the same queue across brokers using Message Groups to
preserve order


> Connection handling is done roughly the same way fanout does. The
> transport is basically a frankenstein of fanout and failover. Once I
> have the functionality complete I think it'll worthwhile to refactor
> the 3 a bit.
>
> Does this sounds pretty reasonable?

Sounds great. I'd be happy for this to replace fanout as its mostly a
case of just choosing the 1 or all brokers that messages go to based
on the type of command & whether or not its in a transaction etc.
Given the different semantics to failover its probably worth keeping
it separate though we could maybe reuse some code at some point.

Its all sounding great so far John! :)

-- 

James
-------
http://radio.weblogs.com/0112098/