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.
1. A computer-executed method comprising:
an inter-stream router routing messages from a plurality of input streams to a plurality of output streams;
wherein the plurality of input streams comprises a particular input stream;
wherein the particular input stream comprises a sequence of messages;
the inter-stream router detecting an input heartbeat message in the particular input stream;
in response to detecting the input heartbeat message, the inter-stream router automatically sending a respective output heartbeat message, that corresponds to the input heartbeat message, to each output stream of the plurality of output streams;
determining 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;
in response to determining that the particular output heartbeat message was processed from the particular output stream, recording, in a heartbeat store, information from the particular output heartbeat message in connection with the particular output stream;
after a failure of the inter-stream router:
selecting a target input heartbeat message based on information in the heartbeat store, and
automatically recovering messages, from the particular input stream, that are after the target input heartbeat message in the sequence of messages;
wherein the method is performed by one or more computing devices.
2. The computer-executed method of claim 1 , wherein:
the particular input stream includes a plurality of ordered heartbeat messages that includes the input heartbeat message;
each heartbeat message, of the plurality of ordered heartbeat messages, is associated with a heartbeat identifier that identifies a position, of the respective heartbeat message, within the ordering of the plurality of ordered heartbeat messages; and
the information, in the heartbeat store, that is recorded from the particular output heartbeat message in connection with the particular output stream comprises information identifying the particular input stream, and a particular heartbeat identifier from the input heartbeat message that corresponds to the particular output heartbeat message.
3. The computer-executed method of claim 2 , wherein selecting the target input heartbeat message comprises:
based, at least in part, on heartbeat identifiers stored in the heartbeat store, determining that the input heartbeat message is a latest heartbeat message, from the particular input stream, that is recorded in the heartbeat store in connection with all output streams of the plurality of output streams; and
in response to determining that the input heartbeat message is the latest heartbeat message, from the particular input stream, that is recorded in the heartbeat store in connection with all output streams of the plurality of output streams, selecting the input heartbeat message to be the target input heartbeat message.
4. The computer-executed method of claim 1 further comprising, prior to the inter-stream router detecting the input heartbeat message in the particular input stream:
determining that a predetermined amount of time has past since inserting a heartbeat message into the particular input stream; and
in response to determining that the predetermined amount of time has past since inserting a heartbeat message into the particular input stream:
generating the input heartbeat message; and
inserting the input heartbeat message into the particular input stream at a tail of the particular input stream.
5. The computer-executed method of claim 4 further comprising identifying, within configuration information, the predetermined amount of time based on the predetermined amount of time being associated with the particular input stream in the configuration information.
6. The computer-executed method of claim 1 further comprising:
determining that a second output heartbeat message, that corresponds to a second input heartbeat message from the particular input stream, was processed from the particular output stream; and
in response to determining that the second output heartbeat message was processed from the particular output stream, recording, in the heartbeat store, information from the second output heartbeat message in connection with the particular output stream;
wherein the information from the second output heartbeat message replaces, in the heartbeat store, the information from the particular output heartbeat message.
7. The computer-executed method of claim 1 further comprising:
the inter-stream router forming an output array of messages from messages received from the plurality of input streams;
wherein the output array is for the particular output stream, and includes messages from the plurality of input streams that match criteria associated with the particular output stream;
wherein the output array is formed in a manner that preserves the ordering of the messages in the output array with respect to the respective ordering of each of the plurality of input streams; and
prior to determining that the particular output heartbeat message was processed from the particular output stream:
adding messages, including the particular output heartbeat message, from a head of the output array to a batch, and
processing the batch as part of the particular output stream.
8. The computer-executed method of claim 7 wherein adding messages, including the particular output heartbeat message, from the head of the output array to the batch comprises adding messages, from the head of the output array, to the batch until addition of a next message in the output array to the batch would violate a particular batch processing constraint.
9. The computer-executed method of claim 7 wherein determining that the particular output heartbeat message was processed from the particular output stream comprises:
detecting that the particular output heartbeat message was included in the batch; and
after detecting that the particular output heartbeat message was included in the batch, detecting that the batch was processed.
10. One or more non-transitory computer-readable media storing one or more sequences of instructions that, when executed by one or more processors, cause:
an inter-stream router detecting an input heartbeat message in a particular input stream that includes a sequence of messages;
wherein the inter-stream router routes messages from a plurality of input streams to a plurality of output streams;
wherein the plurality of input streams comprises the particular input stream;
in response to detecting the input heartbeat message, the inter-stream router automatically sending a respective output heartbeat message, that corresponds to the input heartbeat message, to each output stream of the plurality of output streams;
determining 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;
in response to determining that the particular output heartbeat message was processed from the particular output stream, recording, in a heartbeat store, information from the particular output heartbeat message in connection with the particular output stream;
after a failure of the inter-stream router:
selecting a target input heartbeat message based on information in the heartbeat store, and
automatically recovering messages, from the particular input stream, that are after the target input heartbeat message in the sequence of messages.
11. The one or more non-transitory computer-readable media of claim 10 , wherein:
the particular input stream includes a plurality of ordered heartbeat messages that includes the input heartbeat message;
each heartbeat message, of the plurality of ordered heartbeat messages, is associated with a heartbeat identifier that identifies a position, of the respective heartbeat message, within the ordering of the plurality of ordered heartbeat messages; and
the information, in the heartbeat store, that is recorded from the particular output heartbeat message in connection with the particular output stream comprises information identifying the particular input stream, and a particular heartbeat identifier from the input heartbeat message that corresponds to the particular output heartbeat message.
12. The one or more non-transitory computer-readable media of claim 11 , wherein selecting the target input heartbeat message comprises:
based, at least in part, on heartbeat identifiers stored in the heartbeat store, determining that the input heartbeat message is a latest heartbeat message, from the particular input stream, that is recorded in the heartbeat store in connection with all output streams of the plurality of output streams; and
in response to determining that the input heartbeat message is the latest heartbeat message, from the particular input stream, that is recorded in the heartbeat store in connection with all output streams of the plurality of output streams, selecting the input heartbeat message to be the target input heartbeat message.
13. The one or more non-transitory computer-readable media of claim 10 , wherein the one or more sequences of instructions further comprise instructions that, when executed by one or more processors, cause, prior to the inter-stream router detecting the input heartbeat message in the particular input stream:
determining that a predetermined amount of time has past since inserting a heartbeat message into the particular input stream; and
in response to determining that the predetermined amount of time has past since inserting a heartbeat message into the particular input stream:
generating the input heartbeat message; and
inserting the input heartbeat message into the particular input stream at a tail of the particular input stream.
14. The one or more non-transitory computer-readable media of claim 13 , wherein the one or more sequences of instructions further comprise instructions that, when executed by one or more processors, cause identifying, within configuration information, the predetermined amount of time based on the predetermined amount of time being associated with the particular input stream in the configuration information.
15. The one or more non-transitory computer-readable media of claim 10 , wherein the one or more sequences of instructions further comprise instructions that, when executed by one or more processors, cause:
determining that a second output heartbeat message, that corresponds to a second input heartbeat message from the particular input stream, was processed from the particular output stream; and
in response to determining that the second output heartbeat message was processed from the particular output stream, recording, in the heartbeat store, information from the second output heartbeat message in connection with the particular output stream;
wherein the information from the second output heartbeat message replaces, in the heartbeat store, the information from the particular output heartbeat message.
16. The one or more non-transitory computer-readable media of claim 10 , wherein the one or more sequences of instructions further comprise instructions that, when executed by one or more processors, cause:
the inter-stream router forming an output array of messages from messages received from the plurality of input streams;
wherein the output array is for the particular output stream, and includes messages from the plurality of input streams that match criteria associated with the particular output stream;
wherein the output array is formed in a manner that preserves the ordering of the messages in the output array with respect to the respective ordering of each of the plurality of input streams; and
prior to determining that the particular output heartbeat message was processed from the particular output stream:
adding messages, including the particular output heartbeat message, from a head of the output array to a batch, and
processing the batch as part of the particular output stream.
17. The one or more non-transitory computer-readable media of claim 16 wherein adding messages, including the particular output heartbeat message, from the head of the output array to the batch comprises adding messages, from the head of the output array, to the batch until addition of a next message in the output array to the batch would violate a particular batch processing constraint.
18. The one or more non-transitory computer-readable media of claim 16 wherein determining that the particular output heartbeat message was processed from the particular output stream comprises:
detecting that the particular output heartbeat message was included in the batch; and
after detecting that the particular output heartbeat message was included in the batch, detecting that the batch was processed.