Batch checkpointing for inter-stream messaging system

Embodiments facilitate efficient recovery from an inter-stream messaging system failure by using heartbeat messages (HMs) to act as watermarks for message recovery. Embodiments insert HMs into each of the input streams at configurable regular intervals. The inter-stream router determines that a message being routed is a HM, and corresponding output HMs are generated based on the input HM. Embodiments insert a respective output HM into each of the output streams. Information indicating which output HMs have been processed from the output streams is tracked. After a failure of the inter-stream router, embodiments identify a target HM for each input stream, which is the latest HM sourced from the respective input stream that was processed from all of the output streams. After the inter-stream router restarts, the router initiates message routing, from each input stream, at the location of the respective input stream's target HM within the input stream.

CROSS-REFERENCE TO RELATED APPLICATIONS

FIELD OF THE INVENTION

The present invention relates to inter-stream message routing and, more specifically, to efficiently recovering from failures when messages are batch-routed from multiple incoming streams to multiple outgoing streams.

BACKGROUND

Inter-stream messaging systems, such as those implemented by a Disruptor message router, route messages from one or more input streams to one or more output streams. “Disruptor: High performance alternative to bounded queues for exchanging data between concurrent threads” by Thompson, et al., published May 2011 provides information about the Disruptor router, and is incorporated by reference as if fully set forth herein.

Many times, the messages being routed by an inter-stream messaging system represent events that have been produced by one or more producers. The event-representing messages are added to the input streams by the producers, and are to be processed by one or more consumers that process messages from the output streams. The inter-stream router of such a system routes any given message from an given input stream to one or more of the output streams based on routing criteria for the system. In the case of a Disruptor-implemented inter-stream messaging system, input streams are implemented as partitions of a ring buffer used by the Disruptor router to store messages, and consumers read the messages from the ring buffer partitions.

FIG. 1depicts an example inter-stream messaging system100in which a router application140routes messages from a set of input streams110,120, and130to a set of output streams150,160, and170based on known routing/mixing criteria. In this example system, each input stream is associated with a message producer, and stores event-representing messages from the associated producer. The input streams may be populated by their respective producers at different rates, depending on the functionality of the producers populating those streams.

According to the routing criteria for the example system, router140sorts incoming messages by type, and assigns all messages with a given type to an output stream that is associated with that type. In the depiction ofFIG. 1, input streams110-130include a set of messages, and output streams150-170show the same set of messages having been routed to the various output streams by router140. For example, input stream110includes two withdrawal-type messages112and114, and two deposit-type messages116and118, which were created in response to operations performed by a particular user of the system with which input stream110is associated. The messages in input stream110are ordered from least-recently generated (message112) to most-recently generated (message118).

Output stream150is associated with withdrawal-type messages and output stream170is associated with deposit-type messages. When routing the messages from input stream110, router140sends messages112and114, in that order, to output stream150for processing (see corresponding messages152and156). Router140also sends messages116and118, in that order, to output stream170for processing (see corresponding messages174and178). In this way,FIG. 1depicts the functionality of router140for a given set of messages.

Some systems require strict ordering of routed messages. To illustrate, it shall be assumed that system100is part of a financial system, in which the relative ordering of messages in the input streams must be preserved in the output streams. A financial system is merely one of a virtually unlimited number of types of systems in which the techniques described herein may be used, and the invention is not limited to any particular type of system.

Preserving input ordering of the messages in the output ordering ensures that messages are able to be processed from the output streams in order of message generation, such that the withdrawals, deposits, balance checks, etc., represented by the messages, are processed in order. Thus, the ordering of messages in the input streams ofFIG. 1is preserved in the output streams. For example, message112(“W1”) was input into stream110before message114(“W2”). Accordingly, in output stream150, message152that corresponds to input message112is ordered before message156that corresponds to input message114.

A first message is located “before” a second message in a given stream when the first message is closer to a head of the stream than the second message. Conversely, the second message is located “after” the first message in the stream because the second message is closer to a tail of the stream than the first message. The head of a message stream (or the head of a structure in which messages from the stream are stored) is the portion of the stream with the least-recently inserted messages, and the tail of the stream (or structure) is the portion of the stream with the most-recently inserted messages. To illustrate in the case of output stream150inFIG. 1, message152is at the head of the stream and message158is at the tail of the stream.

Many times, an inter-stream messaging system processes routed messages in batches. Processing messages from an output stream may comprise any kind of processing that logically removes the messages from the stream, such as publishing the messages to a consumer service that takes action based on the messages. When an inter-stream router processes messages in batches, there are times when one or more messages, having been routed to output streams, are not yet processed.

Recovering from router failure scenarios requires determining which messages had been routed but had not yet been processed prior to failure. Efficiently recovering can be difficult. The difficulty in recovering from such a router failure is compounded by the fact that any message can flow to any output stream from any input stream in an inter-stream message system. Thus, there is a possibility of losing messages when the inter-stream router goes down.

One way to facilitate failure recovery of a router failure is to require the inter-stream messaging system to persist, to a persistent store, the entire state of the system including records of messages that have been routed to output streams. However, this solution is inefficient because it requires a significant amount of storage capacity to store the state information, and it takes time and resources to keep a persisted system state up-to-date. Furthermore, the recovery procedure that is based on the persisted state requires expensive querying all of the stored message records, and then finding the messages with minimum/maximum timestamp.

Specifically, in order to perform recovery, generally all stored messages are read from the persisted system state, and the system identifies a minimum timestamp for a particular attribute (such as creation time) across all persisted messages. Messages at the identified minimum timestamp become the starting place for recovery of reading/publishing messages. As discussed above, this process is expensive both in the amount of storage needed and the amount of processing power needed to accomplish recovery. Thus, recovering from an inter-stream router failure based on a stored state of the system is both costly and inefficient.

As such, it would be beneficial to facilitate efficient recovery of an inter-stream messaging system suffering from a router failure, which recovers lost messages while requiring little storage space and low maintenance costs.

DETAILED DESCRIPTION

General Overview

Embodiments facilitate efficient recovery from an inter-stream messaging system failure by using heartbeat messages in input and output streams of the system to act as watermarks for recovery operations. Embodiments accommodate the various speeds at which producers insert messages into the input streams and consumers consume messages from the output streams by identifying an accurate time slice for recovery using the heartbeat message watermarks, which guarantees no loss of messages upon system failure.

Specifically, embodiments insert heartbeat messages into each of the input streams of the inter-stream messaging system at configurable regular intervals. When the inter-stream router encounters a heartbeat message, an output heartbeat message is (a) generated for each output stream based on the input heartbeat message, and (b) inserted into each of the output streams.

A “heartbeat store” is used to store information indicating which output heartbeat messages have been processed from the output streams. After a failure of the router of the inter-stream messaging system, embodiments identify a “target heartbeat message” for each input stream. The “target heartbeat message” of an input stream is the latest heartbeat message from the respective input stream that was processed from all of the output streams.

For example, assume that heartbeat messages A, B and C have been added to input stream1. Further assume that the heartbeat store indicates that all output streams have processed messages A and B, but only some of the output streams have processed message C. In this example, heartbeat message B would be the target heartbeat message for input stream1, since it is the most recent heartbeat from stream1that was processed by all output streams.

The target heartbeat message for a given input stream identifies a point in the input stream before which all messages have been processed. Thus, after the inter-stream router restarts, the router initiates message routing, from each input stream, at the location of the respective input stream's target heartbeat message within the input stream. In the present example, after failure of the inter-stream router, message routing for stream1would begin after heartbeat message B.

This system of inter-stream router failure recovery is very efficient because no querying of messages is required to determine at what location, within the input streams, the router should initiate message routing after restart. Furthermore, the tracking information required to identify target heartbeat messages for the input streams is stored in a very small data structure that is O(N×M), where N is the number of input streams and M is the number of output streams in the system. The size of this data structure is predictable as it is not affected by the volume of data moving through the individual streams.

Furthermore, because the frequency with which heartbeat messages are inserted into each input stream is configurable, the amount of data required to buffer message information that is needed to replay messages upon router restart may be limited according to implementation needs. Thus, the amount of data required to buffer message data may be balanced against the small amount of processing power required to generate and insert heartbeat messages into the input and output streams.

Batching Based on Publication Constraints

Message batching techniques are described hereafter that conform message batches to batch publication constraints and also to message ordering requirements within the batch, without regard to the type of messages being batched or the specifics of the publication constraints being applied. Specifically, an output array of messages is formed from messages received from a plurality of input streams, where the output array is for a particular output stream and includes messages from the plurality of input streams that match criteria associated with the particular output stream. The output array includes multiple messages from each input stream of the plurality of input streams, and within each input stream of the plurality of input streams, the messages are ordered. Accordingly, the output array is formed in a manner that preserves the ordering of the messages with respect to the respective ordering of each of the plurality of input streams.

Messages are added from a head of the output array to a current batch until addition of a next message in the output array to the current batch would violate a particular batch processing constraint imposed on the batch. According to embodiments, one or more additional messages, which follow the message that would cause the constraint violation, are added to the current batch when addition of the one or more additional messages to the batch (a) does not violate the particular batch processing constraint, and (b) continues to preserve the ordering of the messages, in the batch, with respect to the respective ordering of each of the plurality of input streams.

Embodiments may be configured to produce batches based on criteria chosen by a user, including whether a size of the batch is closest to a size constraint on the batch, whether the batch contains the largest possible number of messages, whether the batch has the most even distribution of messages among source input threads, etc.

Heartbeat Messages

FIG. 2depicts a flowchart200for using heartbeat messages in input streams of an inter-stream messaging system to act as watermarks for recovery of messages from a set of output streams after a router failure, according to embodiments. Specifically, at step202, an inter-stream router detects an input heartbeat message in a particular input stream that includes a sequence of messages, where the inter-stream router routes messages from a plurality of input streams to a plurality of output streams, and where the plurality of input streams comprises the particular input stream. For example,FIG. 3depicts an inter-stream messaging system300with many components that correspond to components depicted inFIG. 1, for which the numbering indicated inFIG. 1is preserved. The input and output streams of system300include heartbeat messages302,304,306,308,312A-C,314A-C,316A-C, and318A-C.

According to an embodiment, a recovery application running on a computing device automatically, and periodically, inserts heartbeat messages into each input stream of a given inter-stream messaging system. The recovery application may be router140or may be another distinct application. The frequency with which heartbeat messages are inserted into a given input stream is configurable, e.g., once every minute, once every second, once every two hours, etc. Configurability of inserting heartbeat messages for a given input stream allows for a balance between how much data must be recovered from the input stream upon router failure vs. how much processing power is being used to insert and sample heartbeat messages for the input stream.

Each heartbeat message includes an identifier of the input stream into which the heartbeat message was inserted. This input stream identifier may be included in the message at the time of message generation or insertion into the input stream. Further, the input stream identifier may be inserted into the output heartbeat messages generated by router140, which are then routed to the different output streams, as described in detail below.

Each heartbeat message also includes sequence information that indicates a position of the heartbeat message in the sequence of heartbeat messages for its particular input stream. The sequence information may be a counter value that increases with each heartbeat message for a given stream, or some other sequence indicator, such a system timestamp that indicates the time that the heartbeat message was generated and inserted into the input stream. According to an embodiment, heartbeat messages include a metadata value indicating the time of heartbeat message generation. Such information can indicate, to a system undergoing recovery, how much time was lost when compared to a current system timestamp.

According to an embodiment, heartbeat messages do not carry information produced by a producer, but are used exclusively as watermarks for router failure recovery, as described in detail herein.

An example of the heartbeat message insertion process shall now be given with reference toFIG. 3. Referring toFIG. 3, the recovery application for system300determines that a predetermined amount of time between heartbeat message insertions, for input stream130, has passed since inserting the latest heartbeat message into stream130. In response, the recovery application generates heartbeat message306with a sequence number, “H8”, that uniquely identifies the heartbeat message among all heartbeat messages inserted into input stream130. The recovery application inserts message306into input stream130at the current tail of the input stream, which, at the time, was occupied by message134.

When the predetermined amount of time between heartbeat messages for input stream130passes after inserting message306, during which time a message producer associated with input stream130produced messages136and138, the recovery application generates a heartbeat message308with a sequence number, “H9”. The sequence number of message308indicates that message308is the next heartbeat message after message306, with the identifier “H8”. The recovery application inserts heartbeat message308into input stream130at the current tail of stream130, which, at the time, was occupied by message138. In a similar manner, the recovery application inserts, into input stream120, heartbeat message304, and also inserts, into input stream110, heartbeat message302, i.e., based on configurable periods of time to wait between inserting heartbeat messages for the different input streams.

Because the messages in input streams110-130are ordered, router140routes the messages, from each individual input stream, in the order in which the router pulls messages from the stream. To illustrate for input stream130, router140routes messages132,134,306,136,138, and then308, in that order.

According to one or more embodiments, messages may be in any format. The attributes of a message, such as type, source input stream, message position in an ordering of messages from the source input stream, etc., may be stored in any way, including in the body of the message data, or as metadata for the message, etc.

At step204, in response to detecting the input heartbeat message, the inter-stream router automatically sends a respective output heartbeat message, that corresponds to the input heartbeat message, to each output stream of the plurality of output streams. For example, after routing messages132and134from input stream130, router140pulls heartbeat message306from input stream130. Router140determines that message306is a heartbeat message, e.g., based on metadata in message306that indicates that the message is a heartbeat message. Upon determining that message306is a heartbeat message, router140generates, and sends to each output stream150-170, a respective output heartbeat message that corresponds to heartbeat message306, i.e., heartbeat messages316A-C. Because the messages in the output stream are ordered according to the respective orderings in the source input streams, the output heartbeat messages316A-C (which correspond to heartbeat message306) mark, in each output stream, the same “position” from input stream130.

To illustrate, as shown inFIG. 3, heartbeat messages316A (in output stream150),316B (in output stream160), and316C (in output stream170) all correspond to message306in that they indicate the same combination of heartbeat message identifier (“H8”) and input stream identifier (“130”) as message306. In output stream160, heartbeat message316B is ordered after output message162(which corresponds to input message132), and, in output stream170, heartbeat message316C is ordered after output message172(which corresponds to input message134). Because output stream150does not include any output messages that correspond to messages depicted in input stream130, other than heartbeat messages, the position of heartbeat message316A in output stream150may be placed at any position that precedes heartbeat message318A in the stream. In the example of system300, heartbeat message316A is located at the head of output stream150.

Similarly, router140generates, and sends to all output streams150-170, output heartbeat messages corresponding to each of input heartbeat messages302,304, and308(i.e., heartbeat messages312A-C,314A-C, and318A-C, respectively).

Heartbeat Message Store

At step206, it is determined that a particular output heartbeat message, that corresponds to the input heartbeat message, was processed from a particular output stream of the plurality of output streams. For example,FIG. 4depicts batches402,404, and406of messages that have been processed from each of the output streams150-170, respectively. In the example ofFIG. 4, batch402includes messages314A,318A,154,152, and316A; batch404includes messages166,316B,164, and162; and batch406includes messages176,174,316C, and172. As an example of message processing, the messages in each of batches402-406have been published, by a respective publishing consumer that publishes messages from the associated output stream, to one or more systems that take action based on the messages.

After batches402-406have been processed, the recovery application determines that heartbeat messages314A,318A, and316A-C, from those batches, have been processed. For example, the recovery application comprises a listener that detects which heartbeat messages are included in a given batch, and then detects that the given batch has been processed. In this example embodiment, the recovery application determines that a particular heartbeat message was processed based on the listener detections.

According to an embodiment, a heartbeat message is processed in the same way as messages that are generated by producers that produce substantive messages for the system. In this embodiment, the consumers reject the heartbeat messages being processed from output streams as irrelevant because they do not contain substantive messages for the consumers. The resources required to process heartbeat messages in this way are small, and do not outweigh the storage and processing power gains resulting from failure recovery, according to embodiments. According to an embodiment, heartbeat messages are processed by discarding them upon successful processing of the batch in which the heartbeat messages are included.

At step208, in response to determining that the particular output heartbeat message was processed from the particular output stream, information from the particular output heartbeat message is recorded in a heartbeat store in connection with the particular output stream. For example, when a heartbeat message is processed from a given output stream, the recovery application records information for the heartbeat message, in a heartbeat store, in connection with both the output stream and the input stream that is the source of the heartbeat message.

FIG. 5depicts an example heartbeat store500that reflects the heartbeat messages that have been processed as part of batches402-406indicated inFIG. 4. Specifically, heartbeat store500includes one row for each input stream110-130(i.e., rows502-506), and a column for each output stream150-170(i.e., columns512-516). The cell corresponding to a given row/column combination indicates the latest heartbeat message, from the indicated input stream, which has been processed from the indicated output stream.

As indicated inFIG. 4, none of heartbeat messages312A-C (“H5”), corresponding to heartbeat message302from input stream110, have been processed. Thus, row502, corresponding to input stream110, indicates that the heartbeat message that was inserted into stream110before heartbeat message302, i.e., “H4” (not indicated inFIG. 4), was the latest heartbeat message from stream110to be processed from each of the output streams150-170.

Furthermore, since message314A (“H20”) was processed from output stream150, column512of row504indicates that “H20” is the latest heartbeat message, sourced from input stream120, that was processed from output stream150. However, messages314B and314C have not yet been processed from output streams160and170. Thus, columns514and516of row504indicate that the heartbeat message that was inserted into stream120before heartbeat message304, i.e., “H19” (not shown inFIG. 4), was the latest heartbeat message, sourced from stream120, to be processed from output streams160and170.

From input stream130, heartbeat messages316A,316B,316C, and318A were all processed from output streams150-170in batches402-406indicated inFIG. 4. Though both messages316A and318A were processed from output stream150, only the latest heartbeat message from input stream130processed from output stream150is indicated in heartbeat store500, i.e., “H9” in column512of row506. Columns514and516of row506indicate that “H8” is the latest heartbeat message (i.e., messages318B and318C) sourced from input stream130to be processed from output streams160and170.

Recovery from Inter-stream Router Failure

While routing messages between input and output streams, an inter-stream router may fail, which causes the loss of any message that are not part of a message batch that has been processed from the output streams. For example, after the messages of batches402-406, indicated inFIG. 4, have been processed from output streams150-170, and while the unprocessed messages indicated in output streams150-170inFIG. 4are associated with those streams, router140experiences a failure that requires restarting the router. Upon restarting the router, all information maintained by router140about which messages have been processed, and which messages have been allocated to particular output streams, is lost.

According to an embodiment, steps210and212are performed after a failure of the inter-stream router (e.g., router140). At step210, a target input heartbeat message is selected based on information in the heartbeat store. For example, after router140has had a failure, the recovery application selects a target heartbeat message for each input stream of system300based on the information in heartbeat store500. The target heartbeat message for each input stream provides a watermark for recovering messages that were lost and not published from the input stream.

Selecting the target input heartbeat message for a given input stream comprises the recovery application automatically identifying the latest heartbeat message, from the input stream, that is recorded in the heartbeat store for all output streams of the plurality of output streams. To illustrate, heartbeat store500ofFIG. 5indicates that, for input stream130, heartbeat message “H9” was processed from output stream150, but the latest heartbeat message processed from output streams160and170was “H8”. Thus, the recovery application determines that the target heartbeat message for input stream130is “H8”, because it is the latest heartbeat message that was processed from all of the output streams. The heartbeat message “H9” is not selected as the target heartbeat message for input stream130because the output heartbeat messages for “H9” have not yet been processed from all of the output streams, as indicated in heartbeat store500. In a similar manner, the recovery application determines that the target heartbeat message for input stream110is “H4”, and that the target heartbeat message for input stream120is “H19”.

The target heartbeat message for a given input stream indicates a point within the input stream before which all messages have been processed from all output streams. As such, the recovery application need only cause router140to initiate routing messages from the given input stream starting immediately after the target heartbeat message. Accordingly, at step212, messages are automatically recovered, from the particular input stream, that are after the target input heartbeat message in the sequence of messages. For example, after router140restarts, router140initiates routing messages, for input stream130, from heartbeat message308(“H8”). Similarly, after router140restarts, router140initiates routing messages, for input stream110, from heartbeat message “H4”, and, for input stream120, from heartbeat message “H19”.

According to an embodiment, in order to maintain information for messages that are potentially needed for recovery, the recovery application stores, for each input stream in a data recovery store, messages that precede the latest heartbeat message of the respective input stream processed from all of the output streams. For example, as router140routes each of the messages in input stream120, the recovery application stores information for each of the routed messages in a data recovery store, including data from messages122,124,304,126, and128. When the recovery application records that heartbeat message “H20” has been processed from all of the output streams150-170, the recovery application automatically discards all message data, from the data recovery store, that records messages that precede heartbeat message304(“H20”) in input stream120. In this case, upon determining that heartbeat message “H20” was processed from all of the output streams, the recovery application discards data for messages122,124, and304from the data recovery store.

Thus, embodiments persist a minimum of data for purposes of failure recovery, since embodiments keep track of which messages would be needed to perform failure recovery at any given time. Further, because the recovery is based on the heartbeat message, there is no need to query the data recovery store in order to determine at what point replay of messages should begin upon failure recovery.

According to embodiments, recovery from router failure using heartbeat message as watermarks may cause some messages to be duplicated. For example, as depicted inFIG. 4, “D4” represented by message176was included in batch406, and was processed as part of that batch. However, the input message136that represents “D4” is ordered, in input stream130, after heartbeat message306(“H8”) and before heartbeat message308(“H9”). Thus, if there is a router failure after batch406is processed, but before heartbeat messages318A-C (“H9”) are all processed from output streams150-170, then a duplicate of message136will be re-routed by router140during failure recovery. However, according to an embodiment, message consumers are tolerant to duplicate messages. Thus, duplicate messages being sent out to the output streams does not adversely affect consumer functionality.

Batching Messages for Publication

According to an embodiment, messages are processed from output streams by publishing the messages, which makes the messages available to one or more consumer services that act on the messages. In this embodiment, messages from each output stream may be published independently of messages from the other output streams, such as by one or more message publication technologies that include Amazon Kinesis, Kafka, socket, API, etc.

Many publishing technologies have some constraints, limitations, and/or preferred configurations that restrict the way that messages are published thereby. For example, many publishing technologies constrain the amount of data that may be published in a given period of time, and also constrain the number of messages that may be published in a given period of time. These are non-limiting examples of publishing constraints, and other kinds of constraints may be imposed.

For example, messages in output stream170are of type JSON, and the messages are processed from output stream170by an API publication service, which utilizes HTTP. The API publication service enforces the following constraints on publication from output stream170: the throughput is constrained to a maximum of 5 MB per minute, and the number of messages being published is constrained to a maximum of 5 posts per minute. As a further example, messages in output stream150are composed of binary data, and the messages are processed from output stream150by an Amazon Kinesis publication service, which utilizes RPC. The Amazon Kinesis publication service enforces the following constraints on publication from output stream150: the throughput is constrained to a maximum of 1 MB per second, and the number of messages being published is constrained to a maximum of 1000 messages per second.

These examples are non-limiting, and any kind of time period or constraint may be applied by publishing services. Constraints may involve restrictions based on one or more of: a time window (of second, minutes, etc.); throughput; a number of writes/reads per time window; data format (XML, JSON, Avro, protocol buffers, etc.); data encoding (such as base64, Avro, etc.); data flow attributes, such as an order of bytes, or content of packets, or headers, etc.

Because of publishing constraints, publishing individual messages may cause a constraint on the number of messages that may be published to toll before the maximum allowed throughput of publishing is achieved, which results in inefficient utilization of available bandwidth. However, such issues resulting from publishing constraints may be managed by batching messages, e.g., batches402-406ofFIG. 4, where each batch of messages is published as a single message thereby causing only one message count for purposes of publication. In this way, the size of batches may be configured to maximize the amount of data being published while avoiding contravening constraints on the number of allowed messages.

Given the benefits of message batching, some publishing technologies, such as Amazon Kinesis, expose libraries that facilitate message batching. However, these batching algorithms organize messages into batches without regard to message ordering. Specifically, the batching technologies maximize throughput by fitting messages into batches based on message size, which generally results in scrambling of the original ordering of the messages. Thus, such batching systems are not utilized in inter-stream messaging systems (such as financial systems) that require ordering of messages. Because it is difficult to fully utilize the allowed bandwidth of publishing systems without batching outgoing messages, systems that require ordered publication of messages generally suffer from inefficient utilization of publication bandwidth.

Preserving the Ordering of Messages in Publication Batches

Thus, embodiments provide a batching system that conforms message batches to batch publication constraints and also to message ordering requirements within the batch, without regard to the type of messages being batched or the specifics of the publication constraints being applied. Specifically,FIG. 6depicts a flowchart600for batching messages in order to efficiently satisfy batch processing constraints as well as message ordering requirements, according to embodiments. Flowchart600is described herein in connection with inter-stream messaging systems, as described above. However, embodiments are applicable to any batching that is required to conform to data flow constraints and also to message ordering requirements.

At step602, an output array of messages is formed from messages received from a plurality of input streams, where the output array is for a particular output stream and includes messages from the plurality of input streams that match criteria associated with the particular output stream, where the output array includes multiple messages from each input stream of the plurality of input streams, where, within each input stream of the plurality of input streams, the messages are ordered, and where the output array is formed in a manner that preserves the ordering of the messages with respect to the respective ordering of each of the plurality of input streams. According to embodiments, an array of messages is implemented in any way, including by a queue data structure, a ring buffer data structure (as is used in the Disruptor router, described in further detail below), an in-memory array data structure, etc.

To illustrate step602,FIG. 7Adepicts an example output array700that stores messages for an ordered output stream, e.g., which was generated by an inter-stream router such as router140ofFIG. 3. Specifically, according to this example, router140routes messages from input streams A, B, C, and D into the output stream. Input streams A-D are similar in characteristics to input streams110-130described above, and the output stream corresponding to output array700is similar in characteristics to any of output streams150-170described above. According to an embodiment, router140routes messages, from input streams A-D, to output array700based on the type of the messages matching a type associated with the corresponding output stream.

Output array700stores an ordered list of messages that have been routed to the associated output stream, where the ordering in array700preserves the ordering of the messages with respect to ordering of each of input streams A-D. Specifically, as depicted inFIG. 7A, the messages in array700are labeled the respective source input streams (“A”, “B”, “C”, or “D”). Accordingly, messages710and714are from input stream A; messages704,706,712, and716are from input stream B; messages702,708, and718are from input stream C; and messages720and722are from input stream D. The messages from each stream are further labeled, as a suffix to the input stream label, with the relative position of the message in the source input stream. To illustrate, messages704,706,712, and716from input stream B were included in the stream in that order (from least-recently inserted to most-recently inserted). This ordering from the source input stream is preserved in array700. As shown, messages from one or more other input streams may be interleaved into the messages from input stream B without affecting the preservation of the ordering of the messages from the source input stream.

At step604, messages are added from a head of the output array to a batch until addition of a next message in the output array to the batch would violate a particular batch processing constraint. For example, the messages in array700are processed by a publishing service that subjects the message publication to multiple processing constraints, including restriction of publication throughput to a maximum of 1 MB of data per second, and restriction of the number of messages being published to a maximum of 5 per second.

A batching application, which may be implemented as part of router140, adds messages from a head740of array700to a batch until a next message in the array would cause the batch to violate a batch processing constraint imposed by the publishing service were the batch to be published. Adding messages from head740of the array maintains the ordering of the messages from the source input streams because the source input stream ordering is preserved in the sequence of messages in the array700.

To illustrate,FIGS. 7A-Bdepict batches730A-C generated from messages in output array700, including a batch730A that includes messages702-708from head740of array700. As indicated inFIG. 7A, messages702and708are 120 KB in size, message704is 50 KB in size, and message706is 500 KB in size, which gives a total of 790 KB in batch730A. The next message in array700, message710, is not included in batch730A because inclusion of message710would cause the size of the batch to exceed 1 MB.

At step606, the batch is processed as part of the particular output stream. For example, the batching application determines that one or more applicable publishing constraints allow publication of additional data. To illustrate given the example constraint of 1 MB per second, the batching application determines that it has been at least one second since publication of a previous batch of messages that was large enough to constrain further data publication for one second. In response to determining that a publication of 1 MB of data is allowable based on the applicable publication constraints, the batching application causes batch730A to be published, as a single message, by the publishing service without addition of any other message to the batch.

In this embodiment of batch construction, messages710-722in array700are processed in a different batch. While batch730A could include additional messages from array700while still conforming to the batch processing constraint, it may be advantageous to process batch730A without adding further messages. For example, pulling messages from the head of the output array without further scanning for additional messages is relatively inexpensive. The applicable batch processing constraints allow up to five messages per second, and it may be most efficient to generate multiple (five or less) batches that jointly utilize the maximum throughput, where each batch is constructed using the least amount of processing power necessary to form a batch.

Scanning for Additional Messages to Include in a Batch

According to another embodiment, once the batching application determines that the addition of message710to batch730A would violate the batch processing constraint of 1 MB, the batching application scans for one or more additional messages in output array700, which follow message710, whose addition to the batch (a) would not violate any batch processing constraints, and (b) would continue to preserve the ordering of the messages in the batch with respect to the respective ordering of each of the source input streams.

According to an embodiment, the batching application scans array700to identify additional messages to add to batch730A by evaluating each message subsequent to message710, in order, to determine whether addition of the respective message to the current batch would violate any batch processing constraints. To illustrate, after determining that addition of message710to batch730A would cause the batch to exceed 1 MB in size, the batching application begins evaluation of messages at message712, which is the next message after message710. The batching application determines that addition of message712to the current batch would bring the total size of the batch to 840 KB and, as such, adds message712to the batch (see batch730B).

According to an embodiment, the batching application identifies one or more excluded input streams of the plurality of input streams of the system, where each of the one or more excluded input streams is the source of a message that has been excluded from the batch, and excludes messages from the one or more excluded input streams from the batch. For example, the batching application determines that the next message after message712, message714, is from the same input stream as excluded message710. As such, the batching application automatically excludes message714from the batch.

The batching application then evaluates message716to determine whether addition of the message would cause batch730to exceed 1 MB in size. Because message716is 120 KB, and addition of message716to the current batch would bring the total size of the batch to 960 KB, the batching application adds message716to the batch (see batch730B).

Furthermore, because no other messages in output array700are small enough (40 KB or less) to add to batch730B without violating the batch processing constraint, the batching application adds no additional messages to batch730B prior to the batch being processed. Messages710,714,718,720, and722are left in array700to be processed as part of a different batch. Specifically, after processing of batch730B, output array700includes, at head740, message710, then message714, then message718, and then messages720and722. After these messages, array700would include any further messages routed to the corresponding output stream.

Sets of Candidate Messages

According to an embodiment, the batching application scans the output array to identify additional messages to add to batch730A by compiling one or more sets of candidate messages, where a set of candidate messages comprises messages from the output stream that immediately follow a message, included in the current batch, according to the ordering of messages of the source input stream. Filling a batch of messages from one or more sets of candidate messages has the advantage of equalizing the processing time given to the input streams in any given batch. This equalization could be helpful in a system with input streams that output similar amounts of data over time, especially where a particular input stream might be disadvantaged by a lag in processing of its messages.

To illustrate,FIG. 7Bdepicts candidate sets of messages750A-B. After compiling a preliminary batch730A as described above, the batching application identifies a first set of candidate messages750A from output array700, each message of which immediately follows a respective message, in batch730A, according to the ordering of messages in the source input stream. Specifically, set of candidate messages750A includes a “next” message from each of input streams B, C, and D, as explained below. No message from input stream A is included in the sets of candidate messages because message710that was excluded from preliminary batch730A was from input stream A.

To determine the “next” message from input stream B, the batching application determines that the “last” message from input stream B in preliminary batch730A is message706(“B_2”), and, as such, the “next” message from input stream B in array700is message712(“B_3”). Thus, the batching application includes message712in the set of candidate messages750A. The batching application determines that the “last” message from input stream C in preliminary batch730A is message708(“C_2”). As such, the “next” message from input stream C in array700is message718(“C_3”). Thus, the batching application includes message718in the set of candidate messages750A. No messages from input stream D have yet been included in preliminary batch730A. As such, the “next” message from input stream D in array700is message720(“D_1), which the batching application also includes in the set of candidate messages750A.

According to an embodiment, the batching application adds as many of the messages in the set of candidate messages750A to the current batch as possible given applicable batch processing constraints. According to the example above, because the total size of the set of candidate messages750A is 150 KB, and the total size of batch730A is 790 KB, the batching application automatically adds all of the set of candidate messages750A to the current batch (as shown in batch730C), which brings the total size of the current batch to 940 KB.

According to an embodiment, the batching application forms a second set of candidate messages750B in a manner similar to the construction of the set of candidate messages750A. Specifically, the set of candidate messages750B includes a “next” message for all input streams that have not yet been excluded from the current batch. Because no messages from the set of candidate messages750A were excluded from the current batch, only messages from input stream A are excluded from the set of candidate messages750B based on exclusion of message710(“A_1”) from preliminary batch730A.

Specifically, the set of candidate messages750B includes message716(“B_4”), and message722(“D_2”). There are no other messages from input stream C in array700, so no message from input stream C is included in the set of candidate messages750B. Because addition of message716(size 120 KB) would cause the batch to exceed 1 MB, the batching application excludes this message from the current batch. However, the batching application determines that message722(size 50 KB) fits in batch730C, and, accordingly, the batching application adds message722to batch730C, bringing the size of batch730C to 990 KB.

According to an embodiment, because no other messages in output array700are small enough (10 KB or less) to add to batch730C without violating the batch processing constraint of 1 MB per second, the batching application adds no more messages to batch730C prior to the batch being processed. Thus, after processing of batch730C, output array700includes, at head740, message710, and then messages714and716. After these messages, array700would include any further messages routed to the corresponding output stream from the input streams.

At times, a set of candidate messages includes multiple messages that could be added to the current batch, but which cannot all be added to the current batch based on processing constraints. According embodiments, the batching application chooses which of the multiple candidate messages, in a set of candidate messages, to add to the current batch based on one or more of: maximizing or minimizing the number of messages in the batch, maximizing the size of the batch within the size constraint, evenly distributing processing of messages among input streams, etc. This criteria is configurable by a user.

Furthermore, according to an embodiment, the batching application ceases scanning output array700for additional messages to add to a current batch upon determining that a minimum amount of time has passed since publishing the allowable amount of data during that amount of time. The current batch is caused to be processed without completing the scan for additional messages in order to utilize the available publication bandwidth as quickly and efficiently as possible. According to an embodiment, the batching application automatically determines the amount of data that may still be processed during the allotted time amount, based on the processing constraints, and creates a second batch with a target size configured to maximize the publication throughput.

According to an embodiment, the batching application runs multiple different kinds of batching techniques on the contents of output array700to determine which batching mechanism is more effective. The criteria for most effective batching mechanism is the mechanism that produces a batch, from the output array, that: has a size that is closest to a size constraint on the batch, contains the most number of messages, has the most even distribution of messages among the input threads, or any combination of these criteria. The batch generated from the most effective batching mechanism is caused to be processed. The criteria for most effective batching mechanism used by the batching application is configurable by a user.

The examples discussed herein are simplified for the sake of explanation. However, any number of input streams may feed into any number of output streams, and the complexity of assembling batches greatly increases as the number and variety of messages increases.

Compression and Checksums

According to an embodiment, the recovery application also pushes another type of heartbeat event, of type TIMEOUT, into one or more of the input streams according to a configurable schedule. According to an embodiment that utilizes a ring buffer, when router140processes a heartbeat event of type TIMEOUT, router140automatically flushes the ring buffer. Heartbeat events that are of the type TIMEOUT allows a complex system to deal with network latency that may cause standard heartbeat messages to arrive at router140with increasing latency.

According to an embodiment, system accuracy is preserved by calculating a checksum, e.g., using the CRC32 algorithm, of event content or batch content to prevent data loss or data corruption. For example, before at the time that any given message is pushed into an input stream, embodiments calculate a checksum on the message data and store the checksum value as a metadata attribute. When the message is consumed from an output stream, the checksum for the message is calculated again, and the resulting value is compared to the checksum value stored as a metadata attribute for the message. This comparison is used as a gauge of data loss from messages inserted into the system.

Multi-Stream Messaging System Overview

An inter-stream messaging system may be implemented by one or more computing devices that facilitate the communication of messages, or message records, between input and output streams. The data structures that support the inter-stream messaging system may be configured in any way. For example, an inter-stream messaging system is implemented by one or more queues, where a plurality of input streams populate the one or more queues, and consumers pull messages for the output streams from one or more partitions/shards of the queues.

As another example, an inter-stream messaging system is implemented by a Disruptor router, which utilizes a ring buffer data structure to store/transmit messages. As described in the document “Disruptor: High performance alternative to bounded queues for exchanging data between concurrent threads” by Thompson, et al., published May 2011. Enterprise standard where performance matters along with concurrency

An application (such as router140, the batching application, or the recovery application) runs on a computing device and comprises a combination of software and allocation of resources from the computing device. Specifically, an application is a combination of integrated software components and an allocation of computational resources, such as memory, and/or processes on the computing device for executing the integrated software components on a processor, the combination of the software and computational resources being dedicated to performing the stated functions of the application.

Hardware Overview

For example,FIG. 8is a block diagram that illustrates a computer system800upon which an embodiment of the invention may be implemented. Computer system800includes a bus802or other communication mechanism for communicating information, and a hardware processor804coupled with bus802for processing information. Hardware processor804may be, for example, a general purpose microprocessor.

Computer system800also includes a main memory806, such as a random access memory (RAM) or other dynamic storage device, coupled to bus802for storing information and instructions to be executed by processor804. Main memory806also may be used for storing temporary variables or other intermediate information during execution of instructions to be executed by processor804. Such instructions, when stored in non-transitory storage media accessible to processor804, render computer system800into a special-purpose machine that is customized to perform the operations specified in the instructions.

Computer system800further includes a read only memory (ROM)808or other static storage device coupled to bus802for storing static information and instructions for processor804. A storage device810, such as a magnetic disk, optical disk, or solid-state drive is provided and coupled to bus802for storing information and instructions.

Cloud Computing