Event batching, output sequencing, and log based state storage in continuous query processing

• US 10,255,141 B2
• Filed: 05/20/2016
• Issued: 04/09/2019
• Est. Priority Date: 10/22/2015
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Park, Hoyong, Bishnoi, Sandeep, Deshmukh, Unmesh Anil, Thukkaram, Prabhu
• Primary Examiner(s): Amoroso, Anthony J
• Application Number:
  US15/161,043
• Publication Number:
  US 20170116089A1
• Time in Patent Office: 1,054 Days
• CPC Class Codes: G06F 11/1438 Restarting or rejuvenating
• CPC Class Codes: G06F 11/1471 involving logging of persistent data for recovery
• CPC Class Codes: G06F 16/128 Details of file system snapshots on the file-level, e.g. snapshot creation, administration, deletion use of snapshots for error detection or correction G06F11/14, G06F11/16
• CPC Class Codes: G06F 16/2433 Query languages
• CPC Class Codes: G06F 2201/805 Real-time
• CPC Class Codes: G06F 2201/84 Using snapshots, i.e. a logical point-in-time copy of the data
• CPC Class Codes: G06F 9/542 Event management; Broadcasting; Multicasting; Notifications

Abstract

An event processing system is disclosed that processes events of an event stream and performs the recovery of events during system failure. The recovery of events is performed by generating output sequence numbers for each deterministic output event generated by the event processing system. In an embodiment, the event processing system determines a current output sequence number of a most recently transmitted output event in a output stream of events, a last output sequence number of an output event corresponding to a most recently processed batch of input events and determines a set of one or more output events of the output stream to be transmitted based on the current sequence number and the last output sequence number. The event processing system then transmits the output events related to the application.

Claims

1. A system comprising:

a memory storing a plurality of instructions; and

one or more processors configured to access the memory, wherein the one or more processors are further configured to execute the plurality of instructions to at least;

receive a continuous input stream of events related to an application;

generate a checkpoint marker event;

determine an interval for inserting the checkpoint marker event, the interval based at least in part on a type of the application, a latency requirement of the application, and a frequency at which events of the input stream of events are received;

insert the checkpoint marker event into the continuous input stream of events at each interval;

generate a set of one or more event batches based at least in part on the checkpoint marker event;

determine a size for a particular event batch of the one or more event batches based at least in part on the interval for inserting the checkpoint marker event;

process the continuous input stream of events to generate an output stream of events related to the application;

determine an output sequence number for an output event in the output stream of events;

transmit the output event comprising the output stream of events;

store the output sequence number of the output event;

while the continuous input stream of events is being processed, receive an indication of failure of the system; and

in response to the indication of failure;

determine a current output sequence number of a most recently transmitted output event in the output stream of events;

determine a last output sequence number of an output event corresponding to a most recently processed batch of the one or more event batches;

determine a set of one or more output events of the output stream to be transmitted based at least in part on the current sequence number and the last output sequence number; and

transmit the set of one or more output events related to the application.

2. The system of claim 1, wherein the one or more processors are further configured to not include one or more output events in the set of one or more output events to be transmitted.

3. The system of claim 1, wherein the one or more processors are further configured to transmit output events from the current output sequence number of the most recently transmitted output event in the output stream of events based at least in part on receiving the indication of failure of the system.

4. The system of claim 1, wherein the checkpoint marker event represents information related to an input event in the continuous input stream of events.

5. The system of claim 1, wherein the one or more processors are further configured to:

process the checkpoint marker event;

transmit an acknowledgement message indicating completion of processing of a first event batch in the set of one or more event batches based at least in part on the processing of the checkpoint marker event; and

generate a snapshot of a current state of the system.

6. The system of claim 5, wherein the snapshot of the current state of the system comprises information related to at least one of an input queue state, an operator state, or an output queue state related to at least the first batch of events.

7. The system of claim 5, wherein the one or more processors are further configured to store the output sequence number of an output event in the output stream of events as part of the snapshot.

8. The system of claim 1, wherein the one or more processors are further configured to execute the plurality of instructions to:

receive, from a primary server, an indication that a first event has been processed;

in response to receiving the indication that the first event has been processed, discard a queued event from the output stream, the queued event corresponding to the first event; and

in response to the indication of failure, determine the set of one or more output events of the output stream to be transmitted, wherein the output stream excludes the discarded queued event.

9. A method comprising:

receiving a continuous input stream of events related to an application;

generating a checkpoint marker event;

determining an interval for inserting the checkpoint marker event, the interval based at least in part on a type of the application, a latency requirement of the application, and a frequency at which events of the input stream of events are received;

inserting the checkpoint marker event into the continuous input stream of events at each interval;

generating a set of one or more event batches based at least in part on the checkpoint marker event;

determining a size for a particular event batch of the one or more event batches based at least in part on the interval for inserting the checkpoint marker event;

processing the continuous input stream of events to generate an output stream of events related to the application;

determining an output sequence number for an output event in the output stream of events;

transmitting the output event of the output stream of events;

storing the output sequence number of the output event;

while the continuous input stream of events is being processed, receiving an indication of failure of the system; and

in response to the indication of failure;

determining a current output sequence number of a most recently transmitted output event in the output stream of events;

determining a last output sequence number of an output event corresponding to a most recently processed batch of the one or more event batches;

determining a set of one or more output events of the output stream to be transmitted based at least in part on the current sequence number and the last output sequence number; and

transmitting the set of one or more output events related to the application.

10. The method of claim 9, further comprising not including one or more output events in the set of one or more output events to be transmitted.

11. The method of claim 9, further comprising transmitting output events from the current output sequence number of the most recently transmitted output event in the output stream of events based at least in part on receiving the indication of failure of the system.

12. The method of claim 9, wherein the checkpoint marker event represents information related to an input event in the continuous input stream of events and wherein a size of an event batch is determined based at least in part on the checkpoint marker event.

13. The method of claim 9, further comprising:

processing the checkpoint marker event;

transmitting an acknowledgement message indicating completion of processing of a first event batch in the set of one or more event batches based at least in part on the processing of the checkpoint marker event; and

generating a snapshot of a current state of the system.

14. The method of claim 9, further comprising:

receiving, from a primary server, an indication that a first event has been processed;

in response to receiving the indication that the first event has been processed, discarding a queued event from the output stream, the queued event corresponding to the first event; and

in response to the indication of failure, determining the set of one or more output events of the output stream to be transmitted, wherein the output stream excludes the discarded queued event.

15. A non-transitory computer-readable storage memory storing a plurality of instructions executable by one or more processors, the plurality of instructions comprising:

instructions that cause the one or more processors to receive a continuous input stream of events related to an application;

instructions that cause the one or more processors to generate a checkpoint marker event;

instructions that cause the one or more processors to determine an interval for inserting the checkpoint marker event, the interval based at least in part on a type of the application and a latency requirement of the application;

instructions that cause the one or more processors to insert the checkpoint marker event into the continuous input stream of events at each interval;

instructions that cause the one or more processors to generate a set of one or more event batches based at least in part on the checkpoint marker event;

instructions that cause the one or more processors to determine a size for a particular event batch of the one or more event batches based at least in part on the interval for inserting the checkpoint marker event;

instructions that cause the one or more processors to process the continuous input stream of events to generate an output stream of events related to the application;

instructions that cause the one or more processors to determine an output sequence number for an output event in the output stream of events;

instructions that cause the one or more processors to transmit the output event of the output stream of events;

instructions that cause the one or more processors to store the output sequence number of the output event;

instructions that cause the one or more processors to receive an indication of failure of the system while the continuous input stream of events is being processed;

instructions that cause the one or more processors to determine a current output sequence number of a most recently transmitted output event in the output stream of events;

instructions that cause the one or more processors to determine a last output sequence number of an output event corresponding to a most recently processed batch of the one or more event batches;

instructions that cause the one or more processors to determine a set of one or more output events of the output stream to be transmitted based at least in part on the current sequence number and the last output sequence number; and

instructions that cause the one or more processors to transmit the set of one or more output events related to the application.

16. The non-transitory computer-readable storage memory of claim 15, further comprising instructions that cause the one the one or more processors to transmit output events from the current output sequence number of the most recently transmitted output event in the output stream of events based at least in part on receiving the indication of failure of the system.

17. The non-transitory computer-readable storage memory of claim 15, wherein the plurality of instructions further comprises:

instructions that cause the one or more processors to receive, from a primary server, an indication that a first event has been processed;

instructions that cause the one or more processors, in response to receiving the indication that the first event has been processed, to discard a queued event from the output stream, the queued event corresponding to the first event; and

instructions that cause the one or more processors, in response to the indication of failure, to determine the set of one or more output events of the output stream to be transmitted, wherein the output stream excludes the discarded queued event.

18. The non-transitory computer-readable storage memory of claim 15, wherein the interval is further based on a frequency at which events of the input stream of events are received.