Recovery strategy for a stream processing system

• US 10,606,711 B2
• Filed: 04/16/2018
• Issued: 03/31/2020
• Est. Priority Date: 09/18/2015
• Status: Active Grant

• Current Assignee: Salesforce.com, Inc.
• Original Assignee: Salesforce.com, Inc.
• Inventors: Bishop, Elden Gregory, Chao, Jeffrey
• Primary Examiner(s): Guyton, Philip
• Application Number:
  US15/954,014
• Publication Number:
  US 20180307571A1
• Time in Patent Office: 715 Days
• CPC Class Codes: G06F 11/14 Error detection or correction of the data by redundancy in operation G06F11/16 takes precedence
• 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 11/202 where processing functionality is redundant redundant communication control functionality G06F11/2005, redundant storage control functionality G06F11/2089
• CPC Class Codes: G06F 11/2035 without idle spare hardware
• CPC Class Codes: G06F 11/2048 where the redundant components share neither address space nor persistent storage
• CPC Class Codes: G06F 2201/84 Using snapshots, i.e. a logical point-in-time copy of the data

Abstract

The technology disclosed relates to discovering multiple previously unknown and undetected technical problems in fault tolerance and data recovery mechanisms of modern stream processing systems. In addition, it relates to providing technical solutions to these previously unknown and undetected problems. In particular, the technology disclosed relates to discovering the problem of modification of batch size of a given batch during its replay after a processing failure. This problem results in over-count when the input during replay is not a superset of the input fed at the original play. Further, the technology disclosed discovers the problem of inaccurate counter updates in replay schemes of modern stream processing systems when one or more keys disappear between a batch'"'"'s first play and its replay. This problem is exacerbated when data in batches is merged or mapped with data from an external data store.

Claims

1. A method comprising:

receiving, at an IoT platform and via one or more input pipelines, data from one or more input data streams and transforming the received data for processing, the received data being a collection of one or more events, the IoT platform supporting processing of data streams and historical batch data and supporting virtual applications for processing data for multiple tenants of the IoT platform, wherein processing of data associated with one tenant is isolated from processing of data from another tenant;

assigning priority levels to the one or more input pipelines;

processing the transformed data in two or more batches according to the priority levels assigned to the one or more input pipelines, and wherein each of the two or more batches are processed on a stage by stage basis according to dependencies defined between stages of batch processing; and

monitoring performance metrics of the IoT platform and one or more applications supported by the IoT platform, the IoT platform the metrics being based at least on one or more application records for the one or more applications supported by the IoT platform, the one or more application records including a plurality of contact records, a plurality of account records, other records, or a combination thereof.

2. The method of claim 1, wherein the one or more pipelines are allocated to a container of the IoT platform and the one or more pipelines have connections to at least one common resource external to the container, the method further comprising:

providing a shared connection between the one or more input pipelines and the common resource external to the container.

3. The method of claim 2, further comprising

multiplexing the one or more pipelines over a plurality of workers in the container.

4. The method of claim 1, further comprising:

defining a computing grid that consumes data from a message bus that queues events from the one or more input data streams in numerous batches including the two or batches;

assigning a unique event offset to each event of the two or more batches; and

before processing in the computing grid, at first batch of the two or more batches from a first partition, identifying a current event offset and an end event offset in the first partition and locking the first batch to include events logged between the current event offset and the end event offset.

5. The method of claim 4, further comprising:

persisting the current event offset and the end event offset;

detecting failed processing of the first batch and responding by reprocessing the first batch in the computing grid, wherein between the processing and the reprocessing, the end event offset is subject to change caused by the message bus queuing new events in the first partition; and

restricting the reprocessing of the first batch to events logged between the current event offset and the end event offset, thereby preventing including of the new events in the first batch.

6. An apparatus comprising, comprising:

a processor,

memory in electronic communication with the processor; and

instructions stored in the memory and executable by the processor to cause the apparatus to;

receive, at an IoT platform and via one or more input pipelines, data from one or more input data streams and transforming the received data for processing, the received data being a collection of one or more events, the IoT platform supporting processing of data streams and historical batch data and supporting virtual applications for processing data for multiple tenants of the IoT platform, wherein processing of data associated with one tenant is isolated from processing of data from another tenant;

assign priority levels to the one or more input pipelines;

process the transformed data in two or more batches according to the priority levels assigned to the one or more input pipelines, and wherein each of the two or more batches are processed on a stage by stage basis according to dependencies defined between stages of batch processing; and

monitor performance metrics of the IoT platform and one or more applications supported by the IoT platform, the IoT platform the metrics being based at least on one or more application records for the one or more applications supported by the IoT platform, the one or more application records including a plurality of contact records, a plurality of account records, other records, or a combination thereof.

7. The apparatus of claim 6, wherein the one or more pipelines are allocated to a container of the IoT platform and the one or more pipelines have connections to at least one common resource external to the container, the instructions are further executable by the processor to cause the apparatus to:

provide a shared connection between the one or more input pipelines and the common resource external to the container.

8. The apparatus of claim 7, wherein the instructions are further executable by the processor to cause the apparatus to:

multiplex the one or more pipelines over a plurality of workers in the container.

9. The apparatus of claim 6, wherein the instructions are further executable by the processor to cause the apparatus to:

define a computing grid that consumes data from a message bus that queues events from the one or more input data streams in numerous batches including the two or batches;

assign a unique event offset to each event of the two or more batches; and

before processing in the computing grid, at first batch of the two or more batches from a first partition, identifying a current event offset and an end event offset in the first partition and locking the first batch to include events logged between the current event offset and the end event offset.

10. The apparatus of claim 9, wherein the instructions are further executable by the processor to cause the apparatus to:

persist the current event offset and the end event offset;

detect failed processing of the first batch and responding by reprocessing the first batch in the computing grid, wherein between the processing and the reprocessing, the end event offset is subject to change caused by the message bus queuing new events in the first partition; and

restrict the reprocessing of the first batch to events logged between the current event offset and the end event offset, thereby preventing including of the new events in the first batch.

11. A non-transitory computer-readable medium storing code, the code comprising instructions executable by a processor to:

receive, at an IoT platform and via one or more input pipelines, data from one or more input data streams and transforming the received data for processing, the received data being a collection of one or more events, the IoT platform supporting processing of data streams and historical batch data and supporting virtual applications for processing data for multiple tenants of the IoT platform, wherein processing of data associated with one tenant is isolated from processing of data from another tenant;

assign priority levels to the one or more input pipelines;

process the transformed data in two or more batches according to the priority levels assigned to the one or more input pipelines, and wherein each of the two or more batches are processed on a stage by stage basis according to dependencies defined between stages of batch processing; and

monitor performance metrics of the IoT platform and one or more applications supported by the IoT platform, the IoT platform the metrics being based at least on one or more application records for the one or more applications supported by the IoT platform, the one or more application records including a plurality of contact records, a plurality of account records, other records, or a combination thereof.

12. The non-transitory computer-readable medium of claim 11, wherein the one or more pipelines are allocated to a container of the IoT platform and the one or more pipelines have connections to at least one common resource external to the container, the instructions are further executable to:

provide a shared connection between the one or more input pipelines and the common resource external to the container.

13. The non-transitory computer-readable medium of claim 12, wherein the instructions are further executable to:

multiplex the one or more pipelines over a plurality of workers in the container.

14. The non-transitory computer-readable medium of claim 11, wherein the instructions are further executable to:

define a computing grid that consumes data from a message bus that queues events from the one or more input data streams in numerous batches including the two or batches;

assign a unique event offset to each event of the two or more batches; and

before processing in the computing grid, at first batch of the two or more batches from a first partition, identifying a current event offset and an end event offset in the first partition and locking the first batch to include events logged between the current event offset and the end event offset.

15. The non-transitory computer-readable medium of claim 14, wherein the instructions are further executable to:

persist the current event offset and the end event offset;

detect failed processing of the first batch and responding by reprocessing the first batch in the computing grid, wherein between the processing and the reprocessing, the end event offset is subject to change caused by the message bus queuing new events in the first partition; and

restrict the reprocessing of the first batch to events logged between the current event offset and the end event offset, thereby preventing including of the new events in the first batch.