Showing posts with label activemq. Show all posts
Showing posts with label activemq. Show all posts

Jul 2, 2012

ActiveMQ Broker Networks - Think, Demand Forwarding Bridge


When we build a mental model of how something works, our initial images are always tainted by our experience of the words used to describe it. Sometimes we have to remold our existing perception of the words based on our new learning context, when the words have multiple meanings, we can often start from the wrong premise.

With this in mind, I want to share the meaning of the words I have used to describe ActiveMQ Broker Networks and provide some context to the words. The hope is that this will help build a model in your mind that is a valid reflection of their reality.


A broker network is two or more brokers that are connected via a "Demand Forwarding Bridge". Lets expand on each word in turn, but in reverse:

Bridge

This part is easy, a bridge it is a link between two brokers. The bridge is identified by the transport url of the remote broker it will connect to. The bridge is realized by a socket connection over which messages will pass for specific destinations. A bridge is initiated on a broker via xml configuration of the form:
 <networkConnector url="scheme://host:port" />

Forwarding

The directionality is implicit in the forwarding. Messages flow in one direction only, from the local broker where the bridge is created, to the remote broker where messages end up. Messages are forwarded, which means they are consumed from the local broker, and sent to the remote broker. This is regular JMS send semantics, the message is acknowledged locally when the send to the remote broker completes. So a message only lives in one broker at a time.

Demand

The bridge is aware of demand. In ActiveMQ, demand for messages comes from consumers. A bridge will only forward messages if it knows that there are consumers for that destination on the remote broker. The trick here is the use of ActiveMQ Advisory messages. When a bridge is started, it registers a  consumer with the remote broker for consumer advisory messages. In this way, the bridge becomes aware of the creation and removal of consumers on the remote broker. The bridge reacts to a new consumer advisory notification by creating a local (proxy) consumer for that destination. When this proxy consumer gets a message dispatch, it responds by forwarding (sending) the message to the remote broker. When the remote consumer disconnects, the corresponding remove advisory notification fires and the local proxy consumer is removed. Messages are no longer forwarded for that destination. In this way, the bridge is led by demand.

Next time you encounter <networkConnector ... /> in xml configuration, think of these three words, Demand, Forwarding, Bridge", then proceed to (re)build your mental model.



Nov 22, 2011

ActiveMQ - multiple kahaDB instances (mKahaDB) helping reduce journal disk usage

The default store implementation in ActiveMQ, KahaDB, uses a journal and index. The journal uses a sequence of append-only files to store messages, acknowledgements and broker events. The index holds references to messages on a per destination basis. Essentially, the index holds the runtime state of the broker, mostly in memory, where as the journal maintains the persistence store of  raw data and events. It is the journal of record in a sense.
Periodically, unreferenced journal files are removed through a garbage collection process, so disk usage is kept in check.
In the main, this scheme works well, however, when multiple destinations on a broker are used in very different ways, it can lead to excessive disk usage by the journal. What follows is some detail on a solution to that problem.

Mixed destination usage; frequent fast tasks vs infrequent slow tasks
Imagine a toy makers order-processing. There are two types or orders, custom and standard. A custom order takes a few days to fulfill, a standard order takes a matter of hours. You can easily imagine two order queues, standard and custom. Now imagine that we only process custom orders once a month but process standard orders all the time. So we expect a large backup of custom orders that is slowly consumed at the start of each month and a steady load on the standard order queue.

What the broker sees
From a broker perspective, in the single shared journal, there will be a batch of journal files that are filled with custom order messages. Subsequent journal files that will have mostly 'standard order' messages and acknowledgements with the odd acknowledgement for a 'custom order' message. The sporadic distribution of  acknowledgements for 'custom orders' in the journal files can be problematic because even when that journal file no longer contains any unacked 'standard order' messages, it must still be retained.

Some background on the need to retain journal files
Journal data files are append only. Both messages and acknowledgements are appended, nothing is deleted from a data file. Journal data files that are unreferenced are periodically removed (or archived). The idea is that the index (JMS destination state) can be recreated in full from the journal at any point in time. Any message without a corresponding acknowledgement is deemed valid.


Referenced journal files

In the simplest case, a journal file is 'referenced' if it contains messages that have not been acknowledged by a consumer. The more subtle case reflects the persistence of acknowledgements (acks). A journal file is 'referenced' if it contains acks for messages in any 'referenced' journal file. This means that we cannot garbage collect a journal file that just contains acks until we can garbage collect all of the journal files that contain the corresponding messages. If we did, in the event of a failure that requires recovery of the index, we would miss some acks and replay messages as duplicates.

Problem
So back to the broker perspective of our toy makers order processing. The first range of journal data files remain till the 'custom orders' queue is depleted. Custom order message acknowledgements get dotted across journal files that result from the enqueue/dequeue of the 'standard orders' queue and the end result is lots of referenced journal files and excessive disk usage.

Solution
Reducing the default journal file size can help in this case, but at the cost of more runtime file IO as messages are distributed across more files. In an ideal world, the 'custom order' queue could be partitioned into its own journal where linear appends of messages and acks would result in a minimal set of journal files in use. Correspondingly, the 'standard order' queue with their short lived messages could share a journal.

With the Mulitple KahaDB persistence adapter, destination partitioning across journals is possible. It provides a neat solution to the scenario described above.
Replacing the default persistence adapter configuration:

<persistenceAdapter>
     <kahaDB directory="${activemq.base}/data/kahadb" />
</persistenceAdapter>

with:

<persistenceAdapter>
    <mKahaDB directory="${activemq.base}/data/kahadb">
      <filteredPersistenceAdapters>
       <filteredKahaDB queue="CustomOrders">
        <persistenceAdapter>
          <kahaDB />
        </persistenceAdapter>
       </filteredKahaDB>
       <filteredKahaDB>
        <persistenceAdapter>
          <kahaDB />
        </persistenceAdapter>
       </filteredKahaDB>
      </filteredPersistenceAdapters>
    </mKahaDB>
</persistenceAdapter>
  
The mKahaDB (m, short for multiple) adapter is a collection of filtered persistence adapters. The filtering reuses the destination policy matching feature to match destinations to persistence adapters. In the case of the above configuration, the 'custom orders' queue will use the first instance of kahaDb and all other destinations will map to the second instance. The second filter is empty, so the default 'match any' wild card is in effect.
This configuration, splitting the destinations based on their usage pattern over time, allows the respective journal files to get reclaimed in a linear fashion as messages are consumed and processed, resulting in minimum disk usage.


Overhead

When transactions span persistence adapters, there is an additional overhead of local two phase commit to ensure both journals are atomically updated. Two phase commit requires that the outcome is persisted so there is an additional disk write required per transaction. This can be avoided by colocating destinations that share transactions in a single kahaDB instance. When transactions access a single persistence adapter or when there are no transactions, there is no additional overhead.


Alternative Use Cases: Relaxed Durability Guarantee

Each nested kahaDB instance is fully configurable so one scenario where the use of different persistence adapters makes sense is where your durability guarantee is weaker for some destinations than others. JMS requires that a write be on disk before a send reply is generated by the broker. To this end, a disk sync is issued by default after every journal write. This default behavior is configurable by the kahaDB attribute enableJournalDiskSyncs. If some destinations don't need this guarantee, they can be assigned to a kahaDB instance that has this option disabled and have their writes return faster, leaving it to the file system to complete the write. Here is an example configuration:


<persistenceAdapter>
    <mkahaDB directory="${activemq.base}/data/kahadb">
      <filteredPersistenceAdapters>
      <filteredKahaDB queue="ImportantStuff">
        <persistenceAdapter>
          <kahaDB />
        </persistenceAadapter>
      </filteredkahadb>
      <filteredkahadb queue="NotSoImportantStuff">
        <persistenceAdapter>
          <kahaDB enableJournalDiskSyncs="false"/>
        </persistenceAdapter>
      </filteredKahaDB>
    </filteredPersistenceAdapters>
  </mKahaDB>
</persistenceAdapter>


Aug 16, 2010

Reminder: JMS is client server infrastructure; update broker ∴ update client

Often we get the following question:
If I upgrade the broker to version 5.x, do I also need to upgrade all my clients?
 The short answer is:
maybe, but error on the safe size and upgrade your clients if it does not cause too much disruption.

For the longer answer there are at least two things to consider:
  • Does the reason for upgrade include the need for fixes that affect client side code? If yes, then obviously update all clients. (Issues of this kind typically focus on some aspect of the JMS API or consumer delivery semantics.)
  • Is there an increment to the openwire protocol version? if so, does it affect me? read on...

Does an update to the openwire protocol version affect me?
The openwire protocol is the set of commands that is used to communicate between an ActiveMQ client and an ActiveMQ broker (and from broker to broker in a cluster scenario). The openwire protocol supports version negotiation such that an old client can negotiate the lowest common version with it's peer and use that version. As a result, in most cases, old clients can work as expected with a newer broker.

There are two potential pitfalls that you should be aware of:
  • fixes/features that depend on the openwire version update.
  • the ever increasing and incomplete version testing matrix.

Fixes or features that depend on the openwire version update
These are typically fixes that require additional information to be passed from the clients to the broker or vice versa. Some examples include the addition of a last delivered sequence id parameter to a consumer close command such that the redelivery count could be more accurately calculated. Another is the addition of a reconnecting flag to a connection command that allows duplicate suppression to be implemented consistently at the transport connection level. In some cases it is not obvious if an issue requires a protocol update without some consideration of the implementation, if in doubt ask on the activemq mailing list.

The ever increasing and incomplete version testing matrix
With every protocol version change, there are new additions to the client/server testing matrix. In ActiveMQ, virtually all tests assume a uniform openwire version, with the exception of a few that validate negotiation. The net result is that validation of the compatibility matrix is largely completed by the community. This works in practice but it is important to be aware of. If you are in doubt as to whether a particular scenarios will work across a broker version mismatch, be sure; ask the computer yourself with a little test.

In summary, If you update the broker, you also need to update the clients; or at least consider it!

Oct 15, 2009

Interpreting the ActiveMQ 5.3.0 SpecJMS2007® Result

Hot on the heels of the latest Apache ActiveMQ release, official SpecJMS2007® Results appear. ActiveMQ does 156 vertical and 60 horizontal. But what does that mean?

Some Background
SpecJMS2007® is a representative, long-running, comparative test. Let me take each of these in turn.

By representative, I mean that it contains a mix of business interactions that utilises point to point (or queue semantics) and publish/subscribe (topic) semantics. The message sizes vary within limits using random generators and there is a mix of non-persistent and persistent messages. All persistent messages are delivered and consumed within transactions. In reality, the interactions are based around a supermarket supply chain application which provides a rich tapestry for realistic actor interplay. Supermarkets querying suppliers, suppliers interacting with distribution centers and throughout, management in headquarters, keeping track of all dealings.

By long running, I mean that the scenario lasts a minimum of 30 minutes, excluding a warm-up period. The verification phase is based on periodic throughput and response time sampling during that period. In this way, the test verifies the sustainable load characteristics of a JMS Broker.

By comparative, I mean that the test artifacts and environment are completely specified such that results are totally reproducible. For example, if the broker implementation is swapped out from the ActiveMQ submission bundle, a comparable result for the same platform can be obtained. The platform (or OS and hardware configuration) must be maintained to produce comparable results. This focuses the comparison on the implementation of the broker, which is the intention.

Explain the Numbers

The numbers are seed values. They provide the base value or multiplier on which subsequent decisions like the number of destinations, quantity of messages etc. are determined for a given test run. An increase in the seed value has a cascade effect on the overall load that is placed on the system. If the seed value is too high, the overall load will result in a failed test. Either because of unacceptable throughput variance or because of response times exceeding predetermined ranges. To pass the test for a given seed value, all response time and throughput expectations must be met.

The vertical and horizontal qualifiers refer to the SpecJMS2007® workload topologies. The topologies are not directly comparable because the seed multipliers have different effects in both topologies.

Vertical
As the seed value increases, the vertical scenario aims to increase the number of messages that are processed for a given number of destinations. So in the supermarket supply chain parlance, this means increasing the quantity and variety of stock that is maintained and the frequency of replenishment of said stock. The number of supermarkets and suppliers etc. remains constant as the base seed value increases. In this way, the ability of the broker to deal with increased load on existing destinations is explored. Another way of looking at this is that the depth of the destinations rather than the number of destinations is increased.

Horizontal
As the seed value increases, the horizontal scenario aims to increase the number of destinations while using a fixed load of messages. This corresponds to adding more supermarkets, suppliers and distribution centers. The quantities of stock and the frequency of replenishment is constant. In this way, the ability of the broker to deal concurrently with large numbers of destinations is explored.

In short, the numbers are meaningless in isolation as they are the units of SpecJMS2007® performance measurement and these units have no real-world corollary. Where they are useful is when used in comparison with another run of the SpecJMS2007® test using a different JMS implementation or with some broker configuration tweak.

For example, there are two platform variants of the results, one with Hyper Threading(HT) enabled and the other with HT disabled. The effect is significant indeed, with the vertical successful seed value going from 138 to 156 and horizontal from 52 to 60. So turn HT on!

Jul 14, 2009

Apache ActiveMQ Out Of Memory!

Apache ActiveMQ is adaptable and configurable. A large part of its popularity is due to its flexibility. However, it comes with a default activemq.xml configuration file that cannot possibly suit everybody's needs. The default configuration is a compromise between memory utilisation, low latency and high throughput, with a smattering of feature demonstrations. In all, it is probably too much for one configuration file, but that is another issue that is in part addressed in version 5.3.0.

With the current defaults, it is relatively easy to push the broker's heap memory utilization past the -Xmx512m heap limit passed to the JVM in the start script. When that happens the broker begins to fail in various places with java.lang.OutOfMemoryError: unable to create new native thread or java.lang.OutOfMemoryError: Java heap space.

What to do?
Well Google is your friend but there is also the ActiveMQ FAQ and particularly the entry that deals with the likely causes and relevant configuration that can alleviate ActiveMQ OutOfMemoryError Exceptions.

In short, the answer is nearly always configuration and the intent is that the OutOfMemory FAQ entry will provide a comprehensive reference for the relevant options. Let it be your first port of call.