Patent Description:
Modern computing systems are typically tasked with solving large computational problems, such as effectively processing large amounts of data. To do so, such systems operate in a distributed system environment. A distributed computing system includes components or nodes that may be located on networked computers, which can communicate and/or coordinate with one another by exchanging messages in order to achieve a common goal. Distributed computing systems are typically characterized by concurrency of components, lack of a global clock, and independent failure of components.

In distributed systems, data can be written to and read from data streams representing an ordered sequence of messages, where data can be distributed among many different computing locations. Distributed systems typically preserve an order of in which data is written to data streams, so that it can be read in the same order. However, conventional distributed computing systems do not provide scalability while preserving message order, especially, when large numbers of reader and/or writer clients are attempting to interact with the system as well as the amount of data to be managed. Document<NPL>] discloses a distributed messaging system named Kafka for collecting and delivering high volumes of log data with log latency. Document<NPL>] gives a short introduction into Kafka. Document <NPL>] is a further documentation of Kafka. Document <CIT> discloses a scheduler configured to process messages of at least one of a plurality of computer systems in a further computer system of the plurality of computer systems. The scheduler receives a first message from one of the plurality of computer systems and registers the first message in a first queue. Then, the scheduler receives a second message from one of the plurality of computer systems. The second message has a logical dependency on the first message. The second message is registered in a second queue and in the first queue, wherein the queue entry of the second message in the first queue occurs after the queue entry of the first message in the first queue. The scheduler identifies the logical dependency of the second message on the first message by the order of the corresponding queue entries in the first queue and finally releases the first message for processing before releasing the second message according to the logical dependency. Document <CIT> discloses a publish/subscribe system containing a plurality of brokers, a plurality of subscribers and plurality of brokers including publisher connecting brokers, intermediate brokers and subscriber connecting brokers. Subscriptions are introduced into the system by the subscribers through associated subscription brokers. New subscriptions are aggregated, assigned a virtual start time and propagated through the system toward the publishers. Each broker maintains subscription information in the form of a directed acyclic graph and a broker vector. Messages are published through the system by the publishers through their associated publisher connecting brokers. Each message is assigned a message vector associating subscriptions to that message. The published messages are routed through the brokers toward the subscribers in accordance with comparisons of message brokers and vector brokers conducted at each broker.

In some implementations, the current subject matter relates to a computer implemented method for scalable processes for write-order preserving data stream consumption. The method can include selecting a data partition in a plurality of data partitions of data stream based on a request received from a client processing node, the plurality of data partitions are distributed among a plurality of broker nodes in a distributed messaging system, identifying a broker node in the plurality of broker nodes hosting the selected data partition, and providing, by the identified broker node, the selected data partition to the client processing node for performing at least one function. The client processing node includes a writer processing node which writes data to the selected data partition. At least one of the selecting, the identifying, and the providing can be performed by at least one processor of at least one computing system. The plurality of data partitions in the data stream is arranged using a rooted acyclic directed graph.

In some implementations, the current subject matter can include one or more of the following optional features.

In some implementations, the client processing node can further include a reader processing node. The reader processing node can read data from the selected data partition.

In some implementations, the identification of the broker node can include communicating the request received from the client processing node to any broker node in the plurality of broker nodes, determining that the broker node receiving the request does not host the selected data partition, and identifying the broker node hosting the selected data partition.

In some implementations, providing of the selected partition can include providing, by the identified broker node, the selected data partition, where the selected data
partition is an existing data partition in the data stream. Alternatively, the identified broker node can create a new data partition and provide the created data partition to the client processing node for performing the at least one function.

In some implementations, the function can include at least one of the following: reading data from the selected data partition and writing data to the selected data partition. In some implementations, data can be read from the selected data partition in accordance with an order in which data was written to the selected data partition.

Such computing systems can be connected and can exchange data and/or commands or other instructions or the like via one or more connections, including but not limited to a connection over a network (e.g., the Internet, a wireless wide area network, a local area network, a wide area network, a wired network, or the like), via a direct connection between one or more of the multiple computing systems, etc..

To address the deficiencies of currently available solutions, one or more implementations of the current subject matter provide for scalable processes for write-order preserving data stream consumption in distributed messaging systems.

<FIG> illustrates an exemplary distributed messaging system <NUM>. The system <NUM> can include a plurality of computing nodes that can be communicatively coupled with one another and can process data streams. An example of a distributed message system is Apache Kafka, as developed by Apache Software Foundation, Forest Hill, Maryland, USA. In distributed messaging systems data can be written to and/or read from data streams, which can represent an ordered sequence of messages. Such data streams can be also referred to as feeds, topics and/or transactions. The data can be distributed across physical locations, e.g., on different computing terminals, servers, databases, etc. in a network, and/or managed by separate processes. A physical and/or logical unit that can have a specific role in the distributed messaging system can be referred to as a node of the distributed messaging system. As shown in <FIG>, the distributed messaging system <NUM> can include a plurality of nodes <NUM>. Further, nodes belonging to an instance of the distributed messaging system can be referred to as a cluster.

In the distributed messaging system, one or more processes, referred to as writers or producers, can write or publish data to a data stream. One or more processes, referred to as readers or consumers, can read data (be subscribed to and process the feed of published messages) from a data stream. Readers and writers can be both referred to as clients. In order to perform read and/or write operations, clients exchange messages with the distributed messaging system.

<FIG> illustrates an exemplary system <NUM>. The system <NUM> can include a plurality of writers <NUM>-<NUM><NUM> and a plurality of readers <NUM>-<NUM><NUM>. The writers <NUM> and readers <NUM> can be communicatively coupled to the distributed messaging system <NUM>. The communications between the clients <NUM>, <NUM> and the system <NUM> can be accomplished using a high-performance, language agnostic transmission control protocol ("TCP"). A distributed messaging system can often guarantee to preserve an order in which data is written to a data stream, which means that readers will receive messages in a data stream in the same order that they were written to the stream by a writer.

<FIG> illustrates an exemplary computing system <NUM> including a distributed messaging system, according to some implementation of the current subject matter. The system <NUM> can include a distributed messaging system <NUM>, a plurality of clients <NUM>, <NUM> (e.g., readers, writers), and brokers <NUM>, <NUM>. The distributed messaging system <NUM> can include brokers or multiple processes <NUM>, <NUM>. Each broker <NUM>, <NUM> can communicate with clients <NUM>, <NUM> (readers and writers) and can be responsible for processing a set of data streams. For example, as shown in <FIG>, the broker <NUM> can be assigned to process data stream <NUM> and can be responsible for communicating with clients <NUM> (two writers and one reader); the broker <NUM> can be assigned to process data stream <NUM> and can be responsible for communicating with clients <NUM> (two readers and one writer). Data streams <NUM>, <NUM> can be assigned to brokers using a variety of ways, utilizing different strategies, etc. In case, an assigned broker is defective and/or becomes defective at system runtime, the assignment of brokers can be altered.

A data stream can be subdivided into a rooted acyclic directed graph of partitions. <FIG> illustrates an exemplary acyclic directed graph <NUM>. An acyclic directed graph can be a finite directed graph without directed cycles. The graph can consist of finitely many vertices and edges. Each edge is directed from one vertex to another, such that there is no way to start at a vertex and follow a consistently-directed sequence of edges that eventually loops back to the same vertex again. The graph <NUM> can include a plurality of vertexes <NUM>-<NUM> connected by edges (shown as arrows). For example, vertex <NUM> can be connected to the vertex <NUM> and vertex <NUM>; vertex <NUM> can be connected to vertex <NUM> and vertex <NUM>; and so on. A vertex and/or collection of vertexes can represent a partition of data.

Each partition can represent an ordered sequence of messages in a data stream. <FIG> illustrates an exemplary partition <NUM> containing an ordered sequence of messages <NUM>, <NUM>,. For each writer, exactly one path starting exists, which can start at the root and can represent partitions that the writer wrote to. This constraint can be enforced by a broker when assigning partitions to writers, e.g., by ensuring that the distance between the next assigned partition and the root is greater than the distance between any previously assigned partition and the root.

The partitions of a data stream can be distributed across brokers in the distributed messaging system. Information about existing data streams and the structure of their partition graph can be maintained by the responsible assigned broker and stored persistently to allow other brokers to take over management if it becomes necessary (e.g., in the event of a failover).

<FIG> illustrates an exemplary process <NUM> for writing data to a data stream using the distributed messaging system shown in <FIG>, according to some implementations of the current subject matter. The process <NUM> can be performed by the writer <NUM> and brokers <NUM>, <NUM>. There can be more than one broker that can be involved in this process. The writer <NUM> can generate and transmit a write request to any given broker <NUM> in the distributed messaging system. The write request message can indicate writer's intent to write to a particular data stream. The system can determine whether the broker is responsible and/or has been assigned to process that particular data stream. In the event that the writer did not address the broker responsible for that data stream (i.e., the broker is not responsible and/or has not been assigned to that data stream), the writer <NUM> can receive a response message from the broker <NUM> indicating that the broker <NUM> is not the correct broker and advising the writer <NUM> to contact another broker <NUM> with this request. In some implementations, the broker <NUM> can automatically redirect the writer <NUM> to responsible broker <NUM>. Once the broker responsible for the data stream has been contacted the broker can return the partition that the writer should write to. The data returned by the broker can include a location of the broker hosting the partition.

Then, the writer <NUM> can write to the partition by sending messages to the broker <NUM>, which is hosting the partition. In some cases, the partition can have a predetermined size limit, which can prevent writing of further messages once the size has been exceeded. If the size of the partition is exceeded, the broker <NUM> can transmit a message to the writer <NUM> indicating that writing to the target partition may not be possible and messages should be written so another partition. Then, the writer <NUM> can contact another responsible broker to find out which partition the writer <NUM> should be writing to.

<FIG> illustrates an exemplary process <NUM> for reading data from a data stream using the distributed messaging system shown in <FIG>, according to some implementations of the current subject matter. The process <NUM> can be performed by a reader <NUM> and brokers <NUM>, <NUM>. There can be more than one broker that can be involved in this process. The reader <NUM> can generate and transmit a read request to any given broker <NUM> in the distributed messaging system. The read request message can indicate reader's desire to read from a particular data stream. The system can determine whether the broker is responsible and/or has been assigned to process that particular data stream. In the event that the reader <NUM> did not address the broker responsible for that data stream (i.e., the broker is not responsible and/or has not been assigned to that data stream), the reader <NUM> can receive a response message from the broker <NUM> indicating that the broker <NUM> is not the correct broker and advising the writer <NUM> to contact another broker <NUM> with this read request. In some implementations, the broker <NUM> can automatically redirect the reader <NUM> to the responsible broker <NUM>. Once the broker responsible for the data stream has been contacted, the broker can return the partition that the reader requested. The data returned by the broker can include a location of the broker hosting the partition. After the complete partition has been read, the reader <NUM> can contact the broker again to obtain the next partition to read from. The process <NUM> can be repeated depending on the broker that the reader as contacted.

In some implementations, the current subject matter system can create partitions dynamically. Each time a writer requests a target partition to write to, the responsible broker can determine whether to create a new partition and/or to choose an existing partition (which might potentially be in use by other writers).

Since the partitions are structured in a directed acyclic graph, the current subject mater system can ensure that readers will read messages of each writer in the exact same order they have been written to the stream. The dynamic creation of partitions can be transparent for the readers and/or writers and can make the current subject matter system highly scalable. This can be due to the fact that many writers and/or readers can, in parallel, access many partitions in the same data stream. Since the partitions are distributed across the cluster, there is an inherent load balancing for read and/or write requests.

In some implementations, the current subject matter can be configured to be implemented in a system <NUM>, as shown in <FIG>. The system <NUM> can include a processor <NUM>, a memory <NUM>, a storage device <NUM>, and an input/output device <NUM>. Each of the components <NUM>, <NUM>, <NUM> and <NUM> can be interconnected using a system bus <NUM>. The processor <NUM> can be configured to process instructions for execution within the system <NUM>. In some implementations, the processor <NUM> can be a single-threaded processor. In alternate implementations, the processor <NUM> can be a multi-threaded processor. The processor <NUM> can be further configured to process instructions stored in the memory <NUM> or on the storage device <NUM>, including receiving or sending information through the input/output device <NUM>. The memory <NUM> can store information within the system <NUM>. In some implementations, the memory <NUM> can be a computer-readable medium. In alternate implementations, the memory <NUM> can be a volatile memory unit. In yet some implementations, the memory <NUM> can be a non-volatile memory unit. The storage device <NUM> can be capable of providing mass storage for the system <NUM>. In some implementations, the storage device <NUM> can be a computer-readable medium. In alternate implementations, the storage device <NUM> can be a floppy disk device, a hard disk device, an optical disk device, a tape device, non-volatile solid state memory, or any other type of storage device. The input/output device <NUM> can be configured to provide input/output operations for the system <NUM>. In some implementations, the input/output device <NUM> can include a keyboard and/or pointing device. In alternate implementations, the input/output device <NUM> can include a display unit for displaying graphical user interfaces.

<FIG> illustrates an exemplary method <NUM> for performing error handling/resolution in computing systems, according to some implementations of the current subject matter. At <NUM>, a data partition in a plurality of data partitions of data stream can be selected based on a request received from a client processing node. The client can be a reader or a writer (as shown in <FIG>). The plurality of data partitions can be distributed among a plurality of broker nodes (e.g., broker nodes <NUM>, <NUM> shown in <FIG>) in a distributed messaging system (e.g., distributed messaging system <NUM> shown in <FIG>). At <NUM>, a broker node in the plurality of broker nodes hosting the selected data partition can be identified. As shown in <FIG> and <FIG>, the correct broker node <NUM> or <NUM> can be identified depending on whether the request is to write to the partition (broker node <NUM>) or to read from the partition (broker node <NUM>). At <NUM>, the identified broker node (e.g., broker node <NUM>, <NUM>) can provide the selected data partition to the client processing node for performing at least one function (e.g., reading from the partition or writing to the partition).

In some implementations, the current subject matter can include one or more of the following optional features. The plurality of data partitions in the data stream can be arranged using a rooted acyclic directed graph (as shown in <FIG>).

In some implementations, the client processing node can includes at least one of the following: a reader processing node and a writer processing node. The reader processing node can read data from the selected data partition. The writer processing node can write data to the selected data partition.

In some implementations, the identification of the broker node can include communicating the request received from the client processing node to any broker node (e.g., broker nodes <NUM>, <NUM> shown in <FIG>, <FIG>, respectively) in the plurality of broker nodes, determining that the broker node receiving the request does not host the selected data partition, and identifying the broker node hosting the selected data partition.

In some implementations, providing of the selected partition can include providing, by the identified broker node, the selected data partition, where the selected data partition is an existing data partition in the data stream. Alternatively, the identified broker node can create a new data partition and provide the created data partition to the client processing node for performing the at least one function.

The systems and methods disclosed herein can be embodied in various forms including, for example, a data processor, such as a computer that also includes a database, digital electronic circuitry, firmware, software, or in combinations of them. Moreover, the above-noted features and other aspects and principles of the present disclosed implementations can be implemented in various environments. Such environments and related applications can be specially constructed for performing the various processes and operations according to the disclosed implementations or they can include a general-purpose computer or computing platform selectively activated or reconfigured by code to provide the necessary functionality. The processes disclosed herein are not inherently related to any particular computer, network, architecture, environment, or other apparatus, and can be implemented by a suitable combination of hardware, software, and/or firmware. For example, various general-purpose machines can be used with programs written in accordance with teachings of the disclosed implementations, or it can be more convenient to construct a specialized apparatus or system to perform the required methods and techniques.

The systems and methods disclosed herein can be implemented as a computer program product, i.e., a computer program tangibly embodied in an information carrier, e.g., in a machine readable storage device or in a propagated signal, for execution by, or to control the operation of, data processing apparatus, e.g., a programmable processor, a computer, or multiple computers.

As used herein, the term "user" can refer to any entity including a person or a computer.

Although ordinal numbers such as first, second, and the like can, in some situations, relate to an order; as used in this document ordinal numbers do not necessarily imply an order. For example, ordinal numbers can be merely used to distinguish one item from another. For example, to distinguish a first event from a second event, but need not imply any chronological ordering or a fixed reference system (such that a first event in one paragraph of the description can be different from a first event in another paragraph of the description).

The foregoing description is intended to illustrate but not to limit the scope of the invention, which is defined by the scope of the appended claims.

These computer programs, which can also be referred to programs, software, software applications, applications, components, or code, include machine instructions for a programmable processor, and can be implemented in a high-level procedural and/or object-oriented programming language, and/or in assembly/machine language. The machine-readable medium can store such machine instructions non-transitorily, such as for example as would a non-transient solid state memory or a magnetic hard drive or any equivalent storage medium.

To provide for interaction with a user, the subject matter described herein can be implemented on a computer having a display device, such as for example a cathode ray tube (CRT) or a liquid crystal display (LCD) monitor for displaying information to the user and a keyboard and a pointing device, such as for example a mouse or a trackball, by which the user can provide input to the computer. For example, feedback provided to the user can be any form of sensory feedback, such as for example visual feedback, auditory feedback, or tactile feedback; and input from the user can be received in any form, including, but not limited to, acoustic, speech, or tactile input.

The subject matter described herein can be implemented in a computing system that includes a back-end component, such as for example one or more data servers, or that includes a middleware component, such as for example one or more application servers, or that includes a front-end component, such as for example one or more client computers having a graphical user interface or a Web browser through which a user can interact with an implementation of the subject matter described herein, or any combination of such back-end, middleware, or front-end components. The components of the system can be interconnected by any form or medium of digital data communication, such as for example a communication network. Examples of communication networks include, but are not limited to, a local area network ("LAN"), a wide area network ("WAN"), and the Internet.

A client and server are generally, but not exclusively, remote from each other and typically interact through a communication network.

Claim 1:
A computer-implemented method, comprising:
selecting a data partition (<NUM>) in a plurality of data partitions (<NUM>) of data stream (<NUM>, <NUM>) based on a request received from a client processing node (<NUM>, <NUM>, <NUM>, <NUM>), each data partition (<NUM>) representing an ordered sequence of messages in the data stream (<NUM>, <NUM>),
wherein the plurality of data partitions (<NUM>) are distributed among a plurality of broker nodes (<NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>) in the distributed messaging system (<NUM>, <NUM>),
wherein the plurality of data partitions (<NUM>) in the data stream (<NUM>, <NUM>) is arranged using a rooted acyclic directed graph (<NUM>) which includes a plurality of vertexes (<NUM>-<NUM>) connected by edges, each edge being directed from one vertex to another, such that there is no way to start at a vertex and follow a consistently-directed sequence of edges that eventually loops back to the same vertex again, and
wherein a vertex (<NUM>-<NUM>) and/or collection of vertexes (<NUM>-<NUM>) represents a partition (<NUM>) of data;
identifying a broker node (<NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>) in the plurality of broker nodes (<NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>) hosting the selected data partition (<NUM>);
providing, by the identified broker node (<NUM>, <NUM>, <NUM>, <NUM>, <NUM>, <NUM>), the selected data partition (<NUM>) to the client processing node (<NUM>, <NUM>, <NUM>, <NUM>) for performing at least one function, wherein the client processing node (<NUM>, <NUM>, <NUM>, <NUM>) includes a writer processing node (<NUM>, <NUM>); and
writing, by the writer processing node (<NUM>, <NUM>), data to the selected data partition (<NUM>);
wherein the selecting, the identifying, and the providing is performed using at least one processor (<NUM>) of at least one computing system (<NUM>, <NUM>, <NUM>).