Repeatable message streams for message queues in distributed systems

• US 7,818,386 B2
• Filed: 07/29/2005
• Issued: 10/19/2010
• Est. Priority Date: 12/30/2004
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Meiyyappan, Krishnan, Mishra, Shailendra, Surlaker, Kapil, Bhatt, Neerja
• Primary Examiner(s): Bruckart; Benjamin R
• Assistant Examiner(s): SMARTH, GERALD A
• Application Number:
  US11/192,830
• Publication Number:
  US 20060168080A1
• Time in Patent Office: 1,908 Days
• Field of Search: 709/206, 709/207, 709/213, 709/226, 709/237
• US Class Current: 709/207
• CPC Class Codes: H04L 1/1874 Buffer management
• CPC Class Codes: H04L 2001/0093 Point-to-multipoint
• CPC Class Codes: H04L 49/90 Buffering arrangements

Abstract

Techniques for propagating messages in a distributed system are provided. A set of messages enqueued in a source queue are sent to one or more destination queues. An acknowledgement is received from each of the one or more destination queues, where the acknowledgement indicates which messages of the set of messages have been consumed at the particular destination queue. Based on the acknowledgements, one or more messages of the set messages are ceased to be maintained in the source queue.

Claims

1. A computer-implemented method for propagating messages in a distributed system, the method comprising:

sending, to a plurality of destination queues, a set of messages that are enqueued in a source queue;

wherein the source queue and each destination queue of the plurality of destination queues are maintained at separate nodes of the distributed system;

wherein the set of messages is an ordered sequence of messages;

before all of the set of messages have been consumed at the plurality of destination queues, receiving, from each destination queue of the plurality of destination queues, an acknowledgement indicating which messages of the set of messages have been consumed at said each destination queue, wherein multiple consumers are configured to consume the set of messages from said each destination queue;

wherein the acknowledgement received from said each destination queue indicates a highest numbered message in said each destination queue that was consumed by all of the multiple consumers that consume messages from said each destination queue;

at a node that maintains the source queue, determining a lowest numbered message of all highest numbered messages that are indicated in acknowledgements received from all of the plurality of destination queues;

wherein determining the lowest numbered message comprises comparing to each other the sequence numbers of all highest numbered messages that are indicated in the acknowledgements received from all of the plurality of destination queues;

based on the acknowledgements received from each destination queue of the plurality of destination queues, ceasing to maintain one or more messages of the set of messages enqueued in the source queue;

wherein ceasing to maintain the one or more messages includes ceasing to maintain any message that is numbered lower than the lowest numbered message;

performing a replay operation to re-send from the source queue, and to re-enqueue in at least one destination queue of the plurality of destination queues, at least one message that is numbered higher than the lowest numbered message;

wherein the replay operation is not performed for any messages from the source queue that are numbered lower than the lowest numbered message;

wherein the method is performed by one or more computer systems.

2. The method of claim 1, wherein the step of ceasing to maintain the one or more messages is performed by ceasing to maintain any message of the set of messages that has been consumed by all consumers, of the plurality of destination queues, for which the set of messages was created.

3. The method of claim 1, wherein:

the one or more messages are first one or more messages; and

the method further comprises determining, based on the acknowledgements received from each destination queue of the plurality of destination queues, second one or more messages of the set of messages that need to continue to be maintained in the source queue.

4. The method of claim 1, wherein the step of receiving the acknowledgement from each destination queue of the plurality of destination queues further comprises storing each acknowledgement, wherein each stored acknowledgement is associated with the particular destination queue from which said each acknowledgement was sent.

5. The method of claim 1, wherein the steps of sending the set of messages to the plurality of destination queues, receiving the acknowledgement from each destination queue of the plurality of destination queues, ceasing to maintain the one or more messages in the source queue, and performing the replay operation are performed by a propagation mechanism that is a consumer relative to the source queue and a publisher relative to the plurality of destination queues.

6. The method of claim 1, wherein:

a plurality of publishers publish messages to the source queue; and

the step of sending the set of messages to the plurality of destination queues comprises sending only those messages that have been enqueued in the source queue by a particular publisher of the plurality of publishers.

7. The method of claim 1, wherein the distributed system is a database management system (DBMS).

8. A computer-readable non-transitory storage medium storing one or more sequences of instructions for propagating messages in a distributed system, which instructions, when executed by one or more processors, cause the one or more processors to perform steps comprising:

sending, to a plurality of destination queues, a set of messages that are enqueued in a source queue;

wherein the source queue and each destination queue of the plurality of destination queues are maintained at separate nodes of the distributed system;

wherein the set of messages is an ordered sequence of messages;

before all of the set of messages have been consumed at the plurality of destination queues, receiving, from each destination queue of the plurality of destination queues, an acknowledgement indicating which messages of the set of messages have been consumed at said each destination queue, wherein multiple consumers are configured to consume the set of messages from said each destination queue;

wherein the acknowledgement received from said each destination queue indicates a highest numbered message in said each destination queue that was consumed by all of the multiple consumers that consume messages from said each destination queue;

at a node that maintains the source queue, determining a lowest numbered message of all highest numbered messages that are indicated in acknowledgements received from all of the plurality of destination queues;

wherein determining the lowest numbered message comprises comparing to each other the sequence numbers of all highest numbered messages that are indicated in the acknowledgements received from all of the plurality of destination queues;

based on the acknowledgements received from each destination queue of the plurality of destination queues, ceasing to maintain one or more messages of the set of messages enqueued in the source queue;

wherein ceasing to maintain the one or more messages includes ceasing to maintain any message that is numbered lower than the lowest numbered message;

performing a replay operation to re-send from the source queue, and to re-enqueue in at least one destination queue of the plurality of destination queues, at least one message that is numbered higher than the lowest numbered message;

wherein the replay operation is not performed for any messages from the source queue that are numbered lower than the lowest numbered message.

9. The computer-readable non-transitory storage medium of claim 8, wherein the instructions that cause ceasing to maintain the one or more messages comprise instructions which, when executed by the one or more processors, cause the one or more processors to perform ceasing to maintain any message of the set of messages that has been consumed by all consumers, of the plurality of destination queues, for which the set of messages was created.

10. The computer-readable non-transitory storage medium of claim 8, wherein:

the one or more messages are first one or more messages; and

the one or more sequences of instructions further comprise instructions which, when executed by the one or more processors, cause the one or more processors to perform determining, based on the acknowledgements received from each destination queue of the plurality of destination queues, second one or more messages of the set of messages that need to continue to be maintained in the source queue.

11. The computer-readable non-transitory storage medium of claim 8, wherein the instructions that cause receiving the acknowledgement from each destination queue of the plurality of destination queues further comprise instructions which, when executed by the one or more processors, cause the one or more processors to perform storing each acknowledgement, wherein each stored acknowledgement is associated with the particular destination queue from which said each acknowledgement was sent.

12. The computer-readable non-transitory storage medium of claim 8, wherein the instructions that cause sending the set of messages to the plurality of destination queues, receiving the acknowledgement from each destination queue of the plurality of destination queues, ceasing to maintain the one or more messages in the source queue, and performing the replay operation, are included in a set of instructions for executing a propagation mechanism that is a consumer relative to the source queue and a publisher relative to the plurality of destination queues.

13. The computer-readable non-transitory storage medium of claim 8, wherein:

a plurality of publishers publish messages to the source queue; and

the instructions that cause sending the set of messages to the plurality of destination queues comprise instructions which, when executed by the one or more processors, cause the one or more processors to perform sending only those messages that have been enqueued in the source queue by a particular publisher of the plurality of publishers.

14. The computer-readable non-transitory storage medium of claim 8, wherein the distributed system is a database management system (DBMS).