Message broker system with parallel persistence
A method and messaging system comprising a master broker computer system, a plurality of slave broker computer systems, and a plurality of event stores, for a broker computer system to change its status from a slave broker computer system to a master broker computer system.
1. A method executed by a new master broker computer system for message synchronization, the method comprising:
storing, in persistent storage of the new master broker computer system, a plurality of message events, each message event comprising a message received from a client producer and uniquely identifying the message with an epoch value associated with a prior master broker computer system and a sequence number associated with the message event, the new master broker computer system previously being a slave broker computer system that changed its status based at least in part on failure of the previous master broker computer system;
identifying, by the new master broker computer system, a base value for the plurality of message events, the base value being a highest one of the sequence numbers of the message events stored by the new master broker computer system;
retrieving, by the new master broker computer system, a set of message events from one or more event stores in a plurality of event stores based at least in part on the base value and the highest sequence numbers of message events that have been persisted on each of the plurality of event stores;
generating, by the new master broker computer system, a message event stream based at least in part on the retrieved set of message events;
identifying, by the new master broker computer system, a maximum contiguous message event (MCM) based at least in part on the message event stream, wherein the MCM event is a message event with a highest sequence number prior to any sequence number gap in the message event stream; and
synchronizing, by the new master broker computer system, a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event.
2. The method of claim 1 , further comprising:
incrementing, by the new master broker computer system, a current epoch value to indicate that the new master broker computer system epoch has been initiated.
3. The method of claim 1 , wherein failure of the previous master broker computer system is detected in response to one or more slave broker computer systems in the plurality of slave broker computer systems not receiving a message from the previous master broker computer system for a predetermined threshold of time.
4. The method of claim 1 , wherein retrieving a set of message events from one or more event stores in a plurality of event stores based at least in part on the base value and the highest sequence numbers of message events that have been persisted on each of the plurality of event stores comprises:
retrieving, from the plurality of event stores, sequence information corresponding to the highest sequence numbers of message events that have been persisted on each of the plurality of event stores;
determining a set of message events to retrieve based at least in part on the base value and the sequence information; and
retrieving the set of message events from the one or more of the plurality of event stores.
5. The method of claim 1 , wherein synchronizing a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event comprises:
identifying a synchronization point based at least in part on a back pointer of the MCM event, the back pointer pointing to a previous stabilized message event, wherein message events are stabilized based at least in part on storage in a quorum of event stores in the plurality of event stores; and
republishing any message events with sequence numbers above that of the synchronization point to each of the plurality of event stores and to the plurality of slave broker computer systems with a new epoch number determined for the new master broker computer system.
6. The method of claim 5 , wherein republishing any message events with sequence numbers above that of the synchronization point further comprises:
regenerating the message events with sequence numbers above the synchronization point by using the new epoch number associated with the master broker computer system and a new synchronization number; and
distributing, in parallel, the regenerated message events to the plurality of slave broker computer systems and the plurality of event stores.
7. The method of claim 5 , wherein synchronizing a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event further comprises:
updating the new master broker computer system by storing any message events with sequence numbers above that of the synchronization point in the persistent storage of the new master broker computer system.
8. A new master broker computer system for message synchronization comprising:
one or more processors; and
one or more memories operatively coupled to at least one of the one or more processors and having instructions stored thereon that, when executed by at least one of the one or more processors, cause at least one of the one or more processors to:
store, in persistent storage of the new master broker computer system, a plurality of message events, each message event comprising a message received from a client producer and uniquely identifying the message with an epoch value associated with a prior master broker computer system and a sequence number associated with the message event, the new master broker computer system previously being a slave broker computer system that changed its status based at least in part on failure of the previous master broker computer system;
identify a base value for the plurality of message events, the base value being a highest one of the sequence numbers of the message events stored by the new master broker computer system;
retrieve a set of message events from one or more event stores in a plurality of event stores based at least in part on the base value and the highest sequence numbers of message events that have been persisted on each of the plurality of event stores;
generate a message event stream based at least in part on the retrieved set of message events;
identify a maximum contiguous message event (MCM) based at least in part on the message event stream, wherein the MCM event is a message event with a highest sequence number prior to any sequence number gap in the message event stream; and
synchronize a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event.
9. The new master broker of claim 8 , wherein at least one of the one or more memories has further instructions stored thereon that, when executed by at least one of the one or more processors, cause at least one of the one or more processors to:
increment a current epoch value to indicate that the new master broker computer system epoch has been initiated.
10. The new master broker of claim 8 , wherein failure of the previous master broker computer system is detected in response to one or more slave broker computer systems in the plurality of slave broker computer systems not receiving a message from the previous master broker computer system for a predetermined threshold of time.
11. The new master broker of claim 8 , wherein the instructions that, when executed by at least one of the one or more processors, cause at least one of the one or more processors to retrieve a set of message events from one or more event stores in a plurality of event stores based at least in part on the base value and the highest sequence numbers of message events that have been persisted on each of the plurality of event stores further cause at least one of the one or more processors to:
retrieve, from the plurality of event stores, sequence information corresponding to the highest sequence numbers of message events that have been persisted on each of the plurality of event stores;
determine a set of message events to retrieve based at least in part on the base value and the sequence information; and
retrieve the set of message events from the one or more of the plurality of event stores.
12. The new master broker of claim 8 , wherein the instructions that, when executed by at least one of the one or more processors, cause at least one of the one or more processors to synchronize a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event further cause at least one of the one or more processors to:
identify a synchronization point based at least in part on a back pointer of the MCM event, the back pointer pointing to a previous stabilized message event, wherein message events are stabilized based at least in part on storage in a quorum of event stores in the plurality of event stores; and
republish any message events with sequence numbers above that of the synchronization point to each of the plurality of event stores and to the plurality of slave broker computer systems with a new epoch number determined for the new master broker computer system.
13. The new master broker of claim 12 , wherein the instructions that, when executed by at least one of the one or more processors, cause at least one of the one or more processors to republish any message events with sequence numbers above that of the synchronization point further cause at least one of the one or more processors to:
regenerate the message events with sequence numbers above the synchronization point by using the new epoch number associated with the master broker computer system and a new synchronization number; and
distribute, in parallel, the regenerated message events to the plurality of slave broker computer systems and the plurality of event stores.
14. The new master broker of claim 12 , wherein the instructions that, when executed by at least one of the one or more processors, cause at least one of the one or more processors to synchronize a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event further cause at least one of the one or more processors to:
update the new master broker computer system by storing any message events with sequence numbers above that of the synchronization point in the persistent storage of the new master broker computer system.
15. At least one non-transitory computer-readable medium storing computer-readable instructions that, when executed by a new master broker computer system, causes the new master broker computer system to:
store, in persistent storage of the new master broker computer system, a plurality of message events, each message event comprising a message received from a client producer and uniquely identifying the message with an epoch value associated with a prior master broker computer system and a sequence number associated with the message event, the new master broker computer system previously being a slave broker computer system that changed its status based at least in part on failure of the previous master broker computer system;
identify a base value for the plurality of message events, the base value being a highest one of the sequence numbers of the message events stored by the new master broker computer system;
retrieve a set of message events from one or more event stores in a plurality of event stores based at least in part on the base value and the highest sequence numbers of message events that have been persisted on each of the plurality of event stores;
generate a message event stream based at least in part on the retrieved set of message events;
identify a maximum contiguous message event (MCM) based at least in part on the message event stream, wherein the MCM event is a message event with a highest sequence number prior to any sequence number gap in the message event stream; and
synchronize a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event.
16. The at least one non-transitory computer-readable medium of claim 15 , further storing computer-readable instructions that, when executed by at least one of the one or more computing devices, cause at least one of the one or more computing devices to:
increment a current epoch value to indicate that the new master broker computer system epoch has been initiated.
17. The at least one non-transitory computer-readable medium of claim 15 , wherein failure of the previous master broker computer system is detected in response to one or more slave broker computer systems in the plurality of slave broker computer systems not receiving a message from the previous master broker computer system for a predetermined threshold of time.
18. The at least one non-transitory computer-readable medium of claim 15 , wherein the instructions that, when executed by at least one of the one or more computing devices, cause at least one of the one or more computing devices to retrieve a set of message events from one or more event stores in a plurality of event stores based at least in part on the base value and the highest sequence numbers of message events that have been persisted on each of the plurality of event stores further cause at least one of the one or more computing devices to:
retrieve, from the plurality of event stores, sequence information corresponding to the highest sequence numbers of message events that have been persisted on each of the plurality of event stores;
determine a set of message events to retrieve based at least in part on the base value and the sequence information; and
retrieve the set of message events from the one or more of the plurality of event stores.
19. The at least one non-transitory computer-readable medium of claim 15 , wherein the instructions that, when executed by at least one of the one or more computing devices, cause at least one of the one or more computing devices to synchronize a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event further cause at least one of the one or more computing devices to:
identify a synchronization point based at least in part on a back pointer of the MCM event, the back pointer pointing to a previous stabilized message event, wherein message events are stabilized based at least in part on storage in a quorum of event stores in the plurality of event stores; and
republish any message events with sequence numbers above that of the synchronization point to each of the plurality of event stores and to the plurality of slave broker computer systems with a new epoch number determined for the new master broker computer system.
20. The at least one non-transitory computer-readable medium of claim 19 , wherein the instructions that, when executed by at least one of the one or more computing devices, cause at least one of the one or more computing devices to republish any message events with sequence numbers above that of the synchronization point further cause at least one of the one or more computing devices to:
regenerate the message events with sequence numbers above the synchronization point by using the new epoch number associated with the master broker computer system and a new synchronization number; and
distribute, in parallel, the regenerated message events to the plurality of slave broker computer systems and the plurality of event stores.
21. The at least one non-transitory computer-readable medium of claim 19 , wherein the instructions that, when executed by at least one of the one or more computing devices, cause at least one of the one or more computing devices to synchronize a plurality of slave broker computer systems and the new master broker computer system to a state of the prior master broker system prior to failure based at least in part on the MCM event further cause at least one of the one or more computing devices to:
update the new master broker computer system by storing any message events with sequence numbers above that of the synchronization point in the persistent storage of the new master broker computer system.