Configuring Queues in a SOW

This section outlines the configuration parameters for queues. AMPS provides three distinct types of queues, each designed for specific use cases: Queue, LocalQueue, and GroupLocalQueue.

Defining Queue Replication Type

Queues are defined within the SOW element using the tags Queue, LocalQueue, or GroupLocalQueue. The tag that encloses a queue definition specifies the queue replication type. Full details can be found in the Queue Replication Types section.

To summarize:

• Queue tag: Defines a distributed queue. The queue may be consumed from any instance hosting the queue, and all instances that accept publishes for queue messages must host the queue.

• LocalQueue tag: Defines a queue limited to the local instance only. This queue is independent of any other instance.

• GroupLocalQueue tag: Defines a queue restricted to a subset of instances in a replicated set, typically all within the same group. This queue is hosted by some, but not all, instances in the replicated fabric.

AMPS accepts QueueDefinition as a synonym for Queue.

Queue Configuration

Described below are the configuration items that apply to queues regardless of whether they are fully distributed, local, or distributed within a replication group. Expand each item for more details.

Name (required)

The name of the queue topic. This is the name that consumers subscribe to.

If no Name is provided, AMPS accepts Topic as a synonym for Name in the Queue definition.

MessageType (required)

The message type of the queue.

UnderlyingTopic

A topic name or regular expression for the topic that contains the messages to capture in the queue. These topics must be recorded in a transaction log, and all must be of the same message type as the queue.

Default: The Name of the queue.

That is, if an UnderlyingTopic is not provided, the UnderlyingTopic defaults to the Name of the queue.

DefaultPublishTarget (required if UnderlyingTopic contains regular expression characters)

The topic to publish to when an application publishes a message to the queue. For simplicity, AMPS allows applications to publish messages to the queue, and for those messages to be routed to an UnderlyingTopic.

The DefaultPublishTarget must be one of the topics included in the queue.

This element is required if the UnderlyingTopic contains regular expression characters. Otherwise, this element is optional and defaults to the UnderlyingTopic.

LeasePeriod

The amount of time that a subscriber has ownership of the message before the message is returned to the queue.

For at-least-once delivery semantics, the consumer must process and acknowledge the message within this lease period, or the message may be provided to another subscriber.

The LeasePeriod is measured from the time that AMPS sends the message to the subscriber. Set the LeasePeriod to account for round trip network latency as well as the expected processing time for the subscribers.

Default: unset (no expiration)

Semantics

The delivery semantics to use for this queue.

Regardless of the delivery semantics, AMPS queues deliver a given message to a single subscriber at a time. When a subscriber fails to process a message (that is, the connection to the subscriber closes before the message is acknowledged, or the lease expires before the message is acknowledged), the semantics specify how the failure is handled.

AMPS supports the following semantics:

• at-least-once - With these semantics, you can guarantee that a message has been processed by at least one subscriber, as explained in the Delivery Semantics section. With this value, a subscriber must explicitly remove the message from the queue once the message is processed. If the subscriber connection closes before acknowledging the message, if the subscriber returns the message to the queue, or the lease expires, the AMPS server can deliver the message again to allow another attempt to process the message.

• at-most-once - With these semantics, AMPS removes the message from the queue immediately when AMPS sends the message. This allows you to guarantee that no more than one subscriber will process the message, even if the subscriber that receives the message fails without acknowledging the message.

Default: at-least-once

MaxBacklog

The maximum number of outstanding, unacknowledged messages in the queue at any one time.

This parameter allows you to set limits on the number of pending messages from the queue overall. When the queue reaches the MaxBacklog, no incoming messages are delivered from the queue until a message is removed from the queue (either by expiring, or being acknowledged by a client).

This parameter allows you to avoid overwhelming clients during periods of heavy activity.

Notice that this does not set a limit of any sort on the capacity of the queue. This parameter allows you to limit the number of messages that the queue will make available to subscribers at a given time, but does not restrict the capacity of the queue to track messages.

This backlog number is applied per instance of the queue. That is, each instance of AMPS that hosts an instance of a replicated queue will deliver messages up to the MaxBacklog if messages are available on that instance.

Default: unset (no limit)

MaxPerSubscriptionBacklog

The maximum number of outstanding, unacknowledged messages in the queue for an individual subscription.

This parameter allows you to avoid overwhelming a single subscriber during a period of heavy activity.

Subscribers can declare the maximum number of messages that the subscription is prepared to lease at a given time. This maximum defaults to 1 when there is no maximum explicitly specified for a subscription. AMPS will lease the number specified in the subscription or the maximum set for the queue, whichever is lower.

Notice that this does not set a limit of any sort on the capacity of the queue. This parameter allows you to limit the number of messages that the queue will make available for a single subscription at a given time, but does not restrict the capacity of the queue to track messages.

Default: 1

Expiration

Sets the queue default value for the length of time an individual message can remain in the queue before AMPS considers the message to be undeliverable.

Messages may expire while a subscriber has a lease on the message. AMPS does not send an additional notification in this case.

It is possible for a message to also have an expiration value assigned. In this case, the ExpirationModel parameter determines whether the queue value or the message value is used for an individual message.

Default: unset (no limit)

ExpirationModel

Manages how AMPS applies the queue expiration period and the message expiration period to determine the expiration for the message.

There are four models:

• queue - Always use the queue expiration value for expiration, regardless of the value on the message.

• latest - Use whichever expiration value is greatest; that is, use the value that will keep the message in the queue longest.

• earliest - Use whichever expiration value is smallest; that is, use the value that will expire the message from the queue earliest.

• default - Use the message expiration if the message expiration is set; otherwise use the queue expiration.

Notice that an ExpirationModel may be set even if there is no Expiration set on the queue. In this case, the queue expiration is treated as though it is "unlimited" -- a message that uses the queue value will never expire, and the queue value will always be greater than a message value.

Default: default

Filter

An AMPS Filter that is applied to the UnderlyingTopic. When a Filter is specified, only messages matching the Filter appear in the queue.

By default, there is no filter and all messages from the UnderlyingTopic are presented in the queue.

RecoveryPoint

This option allows you to specify the point at which AMPS begins reviewing the transaction log to recover the state of the queue when AMPS restarts and there is no existing information about the state of the queue. By default, AMPS reviews the full log to determine the contents and state of the queue.

If AMPS has a record of the last point in the transaction log at which all previous messages are acknowledged (stored in the queues.ack file), and the recovery point specified in the configuration file has not changed, AMPS will recover from that point rather than the configured RecoveryPoint, since using the cached information will reduce recovery time.

The RecoveryPoint can be one of the following:

• epoch - Recovery begins at the beginning of the transaction log.

• now - Recovery begins at the time AMPS starts queue recovery, so only new messages are added to the queue.

• creation - Recovery begins at the time the queue was created.

• AMPS Bookmark - When an AMPS bookmark is provided, AMPS starts recovery at the specified bookmark.

• ISO-8601 Timestamp - When a timestamp is provided, AMPS starts recovery at the specified timestamp. The timestamp must be provided in the format AMPS uses for timestamp bookmarks.

Default: epoch

FairnessModel

AMPS provides different methods to distribute messages across active subscriptions:

• fast - AMPS delivers to the first subscription found that can process the message.

• round-robin - AMPS distributes to the next subscription found that can process the message.

• proportional - AMPS delivers to the subscription with the lowest ratio of active messages to available backlog.

Each instance of AMPS independently manages the fairness model for subscriptions on that instance. Fairness model information is not replicated across instances.

Default: proportional for at-least-once queues, round-robin for at-most-once queues

Leasing

Ownership model for leased messages.

AMPS supports the following models:

• strict - AMPS allows a client to acknowledge (sow_delete) only messages that are leased to the client or currently unleased. If a client acknowledges a message leased to another client, there is no effect.

• sublet - AMPS allows any client to acknowledge any message, regardless of whether another client has a lease on the message.

Default: sublet

MaxDeliveries

Specifies an upper bound to the number of times AMPS may deliver a queue message before automatically expiring it.

For example, if AMPS delivers a message twice and MaxDeliveries is set to 2, the message will be expired if the subscriber disconnects or unsubscribes before acknowledging it.

This counter is reset if the server restarts, and the counter is not replicated to other instances.

Default: No maximum (0).

MaxCancels

Specifies a limit to the number of times a subscriber may cancel a lease on a message before it is expired.

For example, if a message is canceled for the second time and MaxCancels is set to 2, AMPS automatically expires the message instead of returning it to the message queue.

This counter is reset if the server restarts, and the counter is not replicated to other instances.

Default: No maximum (0).

Priority

Specifies the order in which messages will be distributed from the queue. When present, this element constructs a value that specifies the priority for messages in the queue. Higher priority messages are delivered first, regardless of the order in which messages have been published to the queue.

The contents of the element can be either the name of a field in the message or an AMPS expression. Either way, the result is treated as an unsigned long value.

For example, to order message delivery based on the /price field of the message, specify the following element:

<Priority>/price</Priority>

To order message delivery based on the product of the /price field and the /quantity field in the message, specify the following element:

<Priority>/price * /quantity</Priority>

This option cannot be specified on a queue if BarrierExpression is specified.

Default: There is no default for this value. If not specified, the delivery order is the order in which messages were processed by this instance of AMPS.

BarrierExpression

Specifies the filter used to identify a barrier message (synchronization point) for this queue. When a message matches this expression, it will be delivered to all current subscribers on the queue when every previous message in the queue has been acknowledged. This provides a simple, reliable way to synchronize workers.

For example, the following configuration item specifies that any message that contains a non-NULL /isSyncPoint value will be treated as a barrier message.

<BarrierExpression>/isSyncPoint IS NOT NULL</BarrierExpression>

The BarrierExpression filter is evaluated when AMPS adds the message to the in-memory state of the queue, and is not re-evaluated (unless the in-memory state of the queue is rebuilt after an instance restart).

This option cannot be specified on a queue if Priority is specified.

Default: There is no default for this value. If not specified, no messages are considered to be barrier messages.

Managing Queue Metadata

Described below are the configuration items for managing queue metadata. Expand each item for more details.

FileBackedMetadata

Specifies whether AMPS should persist metadata about the queue in the journal directory.

This could reduce the active memory footprint of an AMPS instance in cases where a queue has a large number of messages, but it is not being actively consumed. In cases where the queue has a large number of unacknowledged messages when AMPS is restarted, this may also improve recovery time.

This option can be unset or set to a value of enabled.

Default: unset (which indicates that queue metadata will not be persisted)

TargetQueueDepth

The target number of messages to keep active state on.

Providing this option limits the amount of state that AMPS keeps for unacknowledged messages. In most circumstances, the limit will be the number of messages specified in this parameter. If this instance receives a transfer request for a message that is in the transaction log, but outside the currently active set of messages, AMPS will process the queue until it reaches the message to be transferred.

Messages that are present in the transaction log, but for which AMPS does not currently have active state, are not deliverable from the queue topic, cannot be queried, cannot be deleted using a sow_delete by filter, and so on. They are not yet in the set of messages that AMPS considers to be an active part of the queue.

By default, this option is not set. When unset, all messages in the queue are active.

If FileBackedMetadata is enabled, this option is typically unnecessary. In that case, AMPS will process the full state of the queue, but state that is not currently in use can be paged out to the file.

This option cannot be set if Priority or BarrierExpression is specified.

When this parameter is set, it should be set to a number that accounts for several seconds of traffic to the queue, considering replication and acknowledgment speeds. For example, if a queue is typically processed at a rate of 2500 messages per second within minimal replication delays, a reasonable minimum target value might be 12500 or 15000.

Notice that, if subscribers use content filters to selectively process the queue, and no currently active messages match the subscription, no messages will be delivered (even if there are matching messages in the queue that are not currently active).

It is also important to note that this option may increase CPU cost, latency, and disk activity for queues, since metadata for messages is loaded as messages are processed rather than when messages arrive from a publisher.

Default: unset (no limit)

Minimum: 1000

DeferredAckExpiration

Specifies the amount of time for AMPS to retain information about an acknowledgment (sow_delete) message received when the corresponding message is not in the queue. This can occur during failover, when messages are received over replication, or in cases where an application that uses a publish store has been offline for an extended period of time.

This element is configured as an interval, for example, 15m or 2h.

The default value is generally recommended. However, in cases where an instance may receive large volumes of acknowledgments for messages that are not currently in the queue, and are not expected to arrive, setting this to a lower value may somewhat reduce the memory required for managing these acknowledgments.

Default: 1d

Group Local Queue Configuration

When a queue is defined as a GroupLocalQueuethe following configuration item must be provided in addition to the mandatory elements above. Expand the item for more details.

InitialOwner (required for GroupLocalQueue)

For a group local queue, provides the instance Name of the instance that will own the message when the message is first published.

This configuration element is required for a GroupLocalQueue. It is not supported for fully distributed queues or for local only queues (that is, a queue defined with the Queue or LocalQueue element).

All instances in a replicated system that define this queue must define the same InitialOwner.

Notice that AMPS does not validate that the InitialOwner exists and is reachable through replication from the current instance. When configuring a GroupLocalQueue, take care to validate the name and ensure that a replication path exists to and from that instance.

There is no default for this element.

When a queue is defined as a GroupLocalQueue, the following optional configuration element may be provided. Expand the item for more details.

GroupLocalQueueDomain

For a group local queue, provides a way to allow a queue that is hosted in instances that are in different replication groups to be identified as the same queue and function as a single distributed queue.

By default, a GroupLocalQueue is identified using the queue Name and the Group of the AMPS instance that hosts the queue. Queues with the same identity are considered to be instances of the same distributed queue.

When this element is defined, the GroupLocalQueue is identified using the queue Name and the GroupLocalQueueDomain rather than the Group of the instance that hosts the queue.

This element was introduced in version 5.3.4. Previous versions do not consider the Group of the instance or the GroupLocalQueueDomain: instances of the queue with the same Name are considered to be the same distributed queue in those versions regardless of the Group that hosts the queue.

The following configuration snippet shows one way to configure a queue:

<!-- Notice that the topics to use for the queue (ORDERS_.*) must be
     recorded in a transaction log. -->
<SOW>
    <Queue>
        <Name>MQ</Name>
        <MessageType>json</MessageType>
        <UnderlyingTopic>ORDERS_.*</UnderlyingTopic>
        <DefaultPublishTarget>ORDERS_DIRECT</DefaultPublishTarget>
        <LeasePeriod>60s</LeasePeriod>
        <Expiration>1d</Expiration>
        <MaxBacklog>3</MaxBacklog>
    </Queue>
</SOW>

The following configuration snippet shows the configuration for a GroupLocalQueue. Any instance of the queue in the group named CONSUME_INSTANCES or that uses the GroupLocalQueueDomain tag of CONSUME_INSTANCES will be treated as an instance of the same queue.

<SOW>
    <GroupLocalQueue>
        <Name>GroupQueue</Name>
        <MessageType>json</MessageType>
        <UnderlyingTopic>ORDERS_.*</UnderlyingTopic>
        <DefaultPublishTarget>ORDERS_DIRECT</DefaultPublishTarget>
        <GroupLocalQueueDomain>CONSUME_INSTANCES</GroupLocalQueueDomain>
    </GroupLocalQueue>
</SOW>