102 can send an AlterMirror RPC to the secondary cluster 104 to start the mirror topic (e.g., the secondary topic 110). The AlterMirror RPC will work when the secondary topic 110 is in a PendingMirror state and will fail on the secondary topic 110 in any other state.

Once the request is received on the secondary cluster 104, verifications are performed before the secondary topic 110 is converted into a Mirror state. The verification includes determining whether there exists a topic on a remote cluster (e.g., the primary cluster 102) with the same topic ID as the persisted remote topic ID and whether the remote topic (e.g., the primary topic 108) is in a StoppedMirror state with the same stopped log end offsets as the log end offsets at the local topic (e.g., the secondary topic 110). If either of these checks fails, then the mirror topic is failed. This is because either (1) the remote topic has been deleted or recreated, or (2) there was production on the local topic before it went into a PendingMirror state resulting in the logs having diverged. As a result, the local or secondary topic 110 cannot be safely converted to a mirror topic. In either of these cases, the state will transition to a FailedMirror state. A failover command will then need to be issued on the remote or primary topic 108 to make it writable again.

Once it is verified that the secondary topic 110 can transition from a PendingMirror state to a Mirror state, the activation of the secondary topic 110 to a Mirror state is performed. FIG. 7 is a diagram of this stage of the failback process in which the secondary topic 110 is transitioned into a mirror topic after failback.

It is noted that all the steps relative to activating mirrors using the reverse-and-swap command (e.g., the reverse-and-start-mirror command) are expected to complete in the lifecycle of the request. This means that if the verification check fails at any point, the mirror topic is failed and the secondary topic 110 set to FailedMirror state. An error is then returned if this happens. For reverse-and-start-mirror, a background task running the task for that command will complete once the RPC for activating mirrors completes, whether successful or not (e.g., in error).

Referring back to FIG. 6, with respect to the reverse-and-pause-mirror command, once the mirror on the primary cluster 102 has completed, the primary cluster 102 transmits an AlterMirror RPC to the secondary cluster 106 to pause the mirror (secondary) topic 110.

Once the request is received on the secondary cluster 104, verifications are performed before converting the secondary topic 110 into a PausedMirror state. The verification determines whether there exists a topic on the remote cluster (e.g., primary cluster 102) with the same topic ID as the persisted remote topic ID and whether the remote topic (e.g., primary topic 108) is in a StoppedMirror state with the same stopped log end offsets as the local log end offsets of the local topic (e.g., secondary topic 110). If either of these checks fails, then the mirror topic fails, as either (1) the remote topic has been deleted or recreated, or (2) there was production on the local topic before it went into a PendingMirror state, meaning that the logs have diverged and the local or secondary topic 110 cannot be safely converted to a mirror topic. In either of these cases, the state will transition to a FailedMirror. The user can get out of this state by issuing a failover command on the secondary topic 110 to make it writable again. It is important that these checks are performed before the secondary topic 110 is placed into a PausedMirror state, as when the user eventually executes the resume-mirrors command, it must be certain that the secondary topic 110 is in a safe state to resume mirroring.

Once it is verified that the secondary topic 110 can be safely transitioned to a Mirror state in the future, the reverse-and-pause-mirror command is completed. FIG. 8 is a diagram of this alternative stage of the failback process in which the secondary topic 110 is transitioned to a paused mirror state (PausedMirror). When the user is ready to resume mirroring on the secondary topic 110, the user can issue a Resume-Mirror command on the secondary topic 110 to convert the secondary topic 110 into a Mirror state. This triggers an AlterMirror RPC to make the state change to the active mirror topic. Upon the activation of the mirror, the secondary topic 110 is now in an active mirror state as shown in FIG. 7.

FIG. 9 is a flowchart illustrating operations of a method 900 for performing the planned failover, according to some example embodiments. Operations in the method 900 may be performed by the components in the network environment described above with respect to FIG. 1-FIG. 8. Accordingly, the method 900 is described by way of example with reference to components in the network environment. However, it shall be appreciated that at least some of the operations of the method 900 may be deployed on various other hardware configurations or be performed by similar components. Therefore, the method 900 is not intended to be limited to these components.

In operation 902, a component detects that mirror lag is close to zero at a secondary topic 110. The component can provide a notification to a user (e.g., customer) associated with the data being stored to the primary cluster 102 and being mirrored by the secondary cluster 104.

In response to the lag being close to zero, the applications on the primary cluster 102 are shut down in operation 904. In some embodiments, the user issues one or more commands to shut down the applications on the primary cluster 102.

In operation 906, the reverse-and-swap command is executed. In example embodiments, the reverse-and-swap command is only executed when lag is detected to be near zero. Otherwise, the reverse will be unsuccessful if there is not a complete synchronization of all the data. In some embodiments, the reverse-and-swap command is a reverse-and-start-mirror command. In other embodiments, the reverse-and-swap command is a reverse-and-pause-mirror command which delays the mirroring until a resume-mirror command is issued to the primary cluster 102.

During the execution of the reverse-and-swap command a series of operations are performed on the topics being failover. The series of operations include placing the primary topic 108 (e.g., production topic) and corresponding secondary topic 110 (e.g., mirror topic) in a read-only state to allow the mirror topic to synchronize all data. When mirror lag is zero, a promote command is executed that triggers best effort synchronization of consumer offsets (e.g., best effort attempt at mirror topic metadata synchronization). Once the synchronization is completed or a timeout triggered, the secondary (mirror) topic 110 is placed in a StoppedMirror state.

In operation 908, a determination is made whether the secondary (mirror) topic 110 associated with the command is in a stopped state. If the secondary topic 110 is not stopped, then the method 900 waits and performs the determination of operation 908 a predetermined time later.

Once the secondary topic 110 is in the stopped state, then in operation 910, an application is started on the secondary cluster 104. As a result, producers 112 can write to the secondary topic 110 of the secondary cluster 104.

In operation 912, mirroring is started on the corresponding primary topic 108 in the primary cluster 102. In embodiments where the reverse-and-start-mirror command was executed, the mirroring on the primary topic 108 starts immediately in response to a positive verification check. However, in embodiments where the reverse-and-pause-mirror command was executed, a resume-mirror command needs to be issued before mirroring starts on the primary topic 108.

FIG. 10 is a flowchart illustrating operations of a method 1000 for performing the planned failback, according to some example embodiments. Operations in the method 1000 may be performed by the components in the network environment described above with respect to FIG. 1-FIG. 8. Accordingly, the method 1000 is described by way of example with reference to components in the network environment. However, it shall be appreciated that at least some of the operations of the method 1000 may be deployed on various other hardware configurations or be performed by similar components. Therefore, the method 1000 is not intended to be limited to these components.

In operation 1002, mirroring is started on the primary topic 108 in the primary cluster 102 if it has not already been started during the failover. In some embodiments, the primary topic 108 can start mirroring when the resume-mirror command is executed.

In operation 1004, a component detects that mirror lag is close to zero at the primary topic 108. The component can provide a notification to a user (e.g., customer) associated with the data being stored to the primary and secondary clusters 102 and 104.

In response to the lag being near zero, the applications on the secondary cluster 104 are shut down in operation 1006. In some embodiments, the user issues one or more commands to shut down the applications on the secondary cluster 104.

In operation 1008, the reverse-and-swap command is executed on the primary cluster 102. In example embodiments, the reverse-and-swap command is only executed when lag is detected to be near zero. In some embodiments, the reverse-and-swap command is a reverse-and-start-mirror command. In other embodiments, the reverse-and-swap command is a reverse-and-pause-mirror command which delays the mirroring until a resume-mirror command is issued to the secondary cluster 104.

During the execution of the reverse-and-swap command a series of operations are performed on the topics being failback. The series of operations include placing the secondary topic 110 (e.g., production topic) and the corresponding primary topic 108 (e.g., mirror topic) in a read-only state to allow the mirror topics to synchronize all data. When mirror lag is zero, a promote command is executed that triggers best effort synchronization of consumer offsets (e.g., best effort attempt at mirror topic metadata synchronization). Once the synchronization is completed or a timeout triggered, the primary (mirror) topic 108 is placed in a StoppedMirror state.

In operation 1010, a determination is made whether the primary (mirror) topic 108 has moved to the stopped state. When the primary topic 108 is in a stopped state, it ensures that no new data is being written allowing the components to accurately access and confirm that the mirror lag is zero. If the primary topic 108 is not stopped, then the method 1000 waits a predetermined amount of time and performs the determination of operation 1010 again. However, if the primary topic 108 has stopped, then in operation 1012, an application is started on the primary cluster 102 thus activating the primary topic 108 to be in a production state and resume production operations.

In operation 1014, the corresponding secondary topic 110 in the secondary cluster 104 can be transitioned to a mirror state and start mirroring data from the primary topic 108. In embodiments where the reverse-and-start-mirror command was executed, the mirroring on the corresponding secondary topic 110 starts immediately. However, in embodiments where the reverse-and-pause-mirror command was executed, a resume-mirror command needs to be issued before mirroring starts on the corresponding secondary topic 110.

As discussed in FIG. 9 and FIG. 10, the planned failover and the failback processes are similar in order to ensure that data is not lost and offsets are preserved. Essentially, a same reverse command (e.g., reverse-and-swap command) is given only when mirror lag is near zero that causes the production and mirroring clusters to be reversed once the mirror lag becomes zero.

For simplicity of discussion, example embodiments have been discussed with a single topic in each of the primary and secondary clusters 102 and 104. It is noted that any number of topics can exist in the primary and secondary clusters 102 and 104 and that the operations discussed herein can be applied to one or more of the topics in each of the primary and secondary clusters 102 and 104 (e.g., a subset of topics in each cluster). Thus, for example, a subset of the topics can be involved in a planned failover and the failback. When multiple topics are being failover/failback, each topic performs the above discussed operations separately. Thus, if one topic is slow in reaching the StoppedMirror state, other topics that have already reached that state can be transitioned into a writable state separately. Similarly, individual topics can be transitioned to a Mirror state separately and start mirroring independent of other topics in the same cluster.

FIG. 11 illustrates components of a machine 1100, according to some example embodiments, that is able to read instructions from a machine-storage medium (e.g., a machine-storage device, a non-transitory machine-storage medium, a computer-storage medium, or any suitable combination thereof) and perform any one or more of the methodologies discussed herein. Specifically, FIG. 11 shows a diagrammatic representation of the machine 1100 in the example form of a computer device (e.g., a computer) and within which instructions 1124 (e.g., software, a program, an application, an applet, an app, or other executable code) for causing the machine 1100 to perform any one or more of the methodologies discussed herein may be executed, in whole or in part.

For example, the instructions 1124 may cause the machine 1100 to execute the flow diagram of FIG. 9 and FIG. 10. In one embodiment, the instructions 1124 can transform the general, non-programmed machine 1100 into a particular machine (e.g., specially configured machine) programmed to carry out the described and illustrated functions in the manner described.

In alternative embodiments, the machine 1100 operates as a standalone device or may be connected (e.g., networked) to other machines. In a networked deployment, the machine 1100 may operate in the capacity of a server machine or a client machine in a server-client network environment, or as a peer machine in a peer-to-peer (or distributed) network environment. The machine 1100 may be a server computer, a client computer, a personal computer (PC), a tablet computer, a laptop computer, a netbook, a set-top box (STB), a personal digital assistant (PDA), a cellular telephone, a smartphone, a web appliance, a network router, a network switch, a network bridge, or any machine capable of executing the instructions 1124 (sequentially or otherwise) that specify actions to be taken by that machine. Further, while only a single machine is illustrated, the term “machine” shall also be taken to include a collection of machines that individually or jointly execute the instructions 1124 to perform any one or more of the methodologies discussed herein.

The machine 1100 includes a processor 1102 (e.g., a central processing unit (CPU), a graphics processing unit (GPU), a digital signal processor (DSP), an application specific integrated circuit (ASIC), a radio-frequency integrated circuit (RFIC), or any suitable combination thereof), a main memory 1104, and a static memory 1106, which are configured to communicate with each other via a bus 1108. The processor 1102 may contain microcircuits that are configurable, temporarily or permanently, by some or all of the instructions 1124 such that the processor 1102 is configurable to perform any one or more of the methodologies described herein, in whole or in part. For example, a set of one or more microcircuits of the processor 1102 may be configurable to execute one or more modules (e.g., software modules) described herein.

The machine 1100 may further include a graphics display 1110 (e.g., a plasma display panel (PDP), a light emitting diode (LED) display, a liquid crystal display (LCD), a projector, or a cathode ray tube (CRT), or any other display capable of displaying graphics or video). The machine 1100 may also include an input device 1112 (e.g., a keyboard), a cursor control device 1114 (e.g., a mouse, a touchpad, a trackball, a joystick, a motion sensor, or other pointing instrument), a storage unit 1116, a signal generation device 1118 (e.g., a sound card, an amplifier, a speaker, a headphone jack, or any suitable combination thereof), and a network interface device 1120.

The storage unit 1116 includes a machine-storage medium 1122 (e.g., a tangible machine-storage medium) on which is stored the instructions 1124 (e.g., software) embodying any one or more of the methodologies or functions described herein. The instructions 1124 may also reside, completely or at least partially, within the main memory 1104, within the processor 1102 (e.g., within the processor's cache memory), or both, before or during execution thereof by the machine 1100. Accordingly, the main memory 1104 and the processor 1102 may be considered as machine-storage media (e.g., tangible and non-transitory machine-storage media). The instructions 1124 may be transmitted or received over a network 1126 via the network interface device 1120.

In some example embodiments, the machine 1100 may be a portable computing device and have one or more additional input components (e.g., sensors or gauges). Examples of such input components include an image input component (e.g., one or more cameras), an audio input component (e.g., a microphone), a direction input component (e.g., a compass), a location input component (e.g., a global positioning system (GPS) receiver), an orientation component (e.g., a gyroscope), a motion detection component (e.g., one or more accelerometers), an altitude detection component (e.g., an altimeter), and a gas detection component (e.g., a gas sensor). Inputs harvested by any one or more of these input components may be accessible and available for use by any of the modules described herein.

Executable Instructions and Machine-Storage Medium

The various memories (e.g., 1104, 1106, and/or memory of the processor(s) 1102) and/or storage unit 1116 may store one or more sets of instructions and data structures (e.g., software) 1124 embodying or utilized by any one or more of the methodologies or functions described herein. These instructions, when executed by processor(s) 1102 cause various operations to implement the disclosed embodiments.

As used herein, the terms “machine-storage medium,” “device-storage medium,” “computer-storage medium” (referred to collectively as “machine-storage medium 1122”) mean the same thing and may be used interchangeably in this disclosure. The terms refer to a single or multiple storage devices and/or media (e.g., a centralized or distributed database, and/or associated caches and servers) that store executable instructions and/or data, as well as cloud-based storage systems or storage networks that include multiple storage apparatus or devices. The terms shall accordingly be taken to include, but not be limited to, solid-state memories, and optical and magnetic media, including memory internal or external to processors. Specific examples of machine-storage media, computer-storage media, and/or device-storage media 1122 include non-volatile memory, including by way of example semiconductor memory devices, e.g., erasable programmable read-only memory (EPROM), electrically erasable programmable read-only memory (EEPROM), FPGA, and flash memory devices; magnetic disks such as internal hard disks and removable disks; magneto-optical disks; and CD-ROM and DVD-ROM disks. The terms machine-storage medium or media, computer-storage medium or media, and device-storage medium or media 1122 specifically exclude carrier waves, modulated data signals, and other such media, at least some of which are covered under the term “signal medium” discussed below. In this context, the machine-storage medium is non-transitory.

Signal Medium

The term “signal medium” or “transmission medium” shall be taken to include any form of modulated data signal, carrier wave, and so forth. The term “modulated data signal” means a signal that has one or more of its characteristics set or changed in such a matter as to encode information in the signal.

Computer Readable Medium

The terms “machine-readable medium,” “computer-readable medium” and “device-readable medium” mean the same thing and may be used interchangeably in this disclosure. The terms are defined to include both machine-storage media and signal media. Thus, the terms include both storage devices/media and carrier waves/modulated data signals.

The instructions 1124 may further be transmitted or received over a communications network 1126 using a transmission medium via the network interface device 1120 and utilizing any one of a number of well-known transfer protocols (e.g., HTTP). Examples of communication networks 1126 include a local area network (LAN), a wide area network (WAN), the Internet, mobile telephone networks, plain old telephone service (POTS) networks, and wireless data networks (e.g., Wi-Fi, LTE, and WiMAX networks). The term “transmission medium” shall be taken to include any intangible medium that is capable of storing, encoding, or carrying instructions 1124 for execution by the machine 1100, and includes digital or analog communications signals or other intangible medium to facilitate communication of such software.

Throughout this specification, plural instances may implement components, operations, or structures described as a single instance. Although individual operations of one or more methods are illustrated and described as separate operations, one or more of the individual operations may be performed concurrently, and nothing requires that the operations be performed in the order illustrated. Structures and functionality presented as separate components in example configurations may be implemented as a combined structure or component. Similarly, structures and functionality presented as a single component may be implemented as separate components. These and other variations, modifications, additions, and improvements fall within the scope of the subject matter herein.

Certain embodiments are described herein as including logic or a number of components, modules, or mechanisms. Modules may constitute either software modules (e.g., code embodied on a machine-storage medium or in a transmission signal) or hardware modules. A “hardware module” is a tangible unit capable of performing certain operations and may be configured or arranged in a certain physical manner. In various example embodiments, one or more computer systems (e.g., a standalone computer system, a client computer system, or a server computer system) or one or more hardware modules of a computer system (e.g., a processor or a group of processors) may be configured by software (e.g., an application or application portion) as a hardware module that operates to perform certain operations as described herein.

In some embodiments, a hardware module may be implemented mechanically, electronically, or any suitable combination thereof. For example, a hardware module may include dedicated circuitry or logic that is permanently configured to perform certain operations. For example, a hardware module may be a special-purpose processor, such as a field programmable gate array (FPGA) or an ASIC. A hardware module may also include programmable logic or circuitry that is temporarily configured by software to perform certain operations. For example, a hardware module may include software encompassed within a general-purpose processor or other programmable processor. It will be appreciated that the decision to implement a hardware module mechanically, in dedicated and permanently configured circuitry, or in temporarily configured circuitry (e.g., configured by software) may be driven by cost and time considerations.

Accordingly, the term “hardware module” should be understood to encompass a tangible entity, be that an entity that is physically constructed, permanently configured (e.g., hardwired), or temporarily configured (e.g., programmed) to operate in a certain manner or to perform certain operations described herein. As used herein, “hardware-implemented module” refers to a hardware module. Considering embodiments in which hardware modules are temporarily configured (e.g., programmed), each of the hardware modules need not be configured or instantiated at any one instance in time. For example, where the hardware modules comprise a general-purpose processor configured by software to become a special-purpose processor, the general-purpose processor may be configured as respectively different hardware modules at different times. Software may accordingly configure a processor, for example, to constitute a particular hardware module at one instance of time and to constitute a different hardware module at a different instance of time.

Hardware modules can provide information to, and receive information from, other hardware modules. Accordingly, the described hardware modules may be regarded as being communicatively coupled. Where multiple hardware modules exist contemporaneously, communications may be achieved through signal transmission (e.g., over appropriate circuits and buses) between or among two or more of the hardware modules. In embodiments in which multiple hardware modules are configured or instantiated at different times, communications between such hardware modules may be achieved, for example, through the storage and retrieval of information in memory structures to which the multiple hardware modules have access. For example, one hardware module may perform an operation and store the output of that operation in a memory device to which it is communicatively coupled. A further hardware module may then, at a later time, access the memory device to retrieve and process the stored output. Hardware modules may also initiate communications with input or output devices, and can operate on a resource (e.g., a collection of information).

The various operations of example methods described herein may be performed, at least partially, by one or more processors that are temporarily configured (e.g., by software) or permanently configured to perform the relevant operations. Whether temporarily or permanently configured, such processors may constitute processor-implemented modules that operate to perform one or more operations or functions described herein. As used herein, “processor-implemented module” refers to a hardware module implemented using one or more processors.

Similarly, the methods described herein may be at least partially processor-implemented, a processor being an example of hardware. For example, at least some of the operations of a method may be performed by one or more processors or processor-implemented modules. Moreover, the one or more processors may also operate to support performance of the relevant operations in a “cloud computing” environment or as a “software as a service” (SaaS). For example, at least some of the operations may be performed by a group of computers (as examples of machines including processors), with these operations being accessible via a network (e.g., the Internet) and via one or more appropriate interfaces (e.g., an application program interface (API)).

The performance of certain of the operations may be distributed among the one or more processors, not only residing within a single machine, but deployed across a number of machines. In some example embodiments, the one or more processors or processor-implemented modules may be located in a single geographic location (e.g., within a home environment, an office environment, or a server farm). In other example embodiments, the one or more processors or processor-implemented modules may be distributed across a number of geographic locations.

In view of the disclosure above, various examples are set forth below. It should be noted that one or more features of an example, taken in isolation or combination, should be considered within the disclosure of this application

EXAMPLES

Example 1 is a method for seamless failback after a planned failover. The method comprises executing a reverse-and-swap command to transition a primary topic on a primary cluster from a mirror state to a production state and a corresponding secondary topic on a secondary cluster from the production state to the mirror state; based on the reverse-and-swap command, placing the primary topic and the corresponding secondary topic in a read-only state to allow the primary topic to synchronize all data; detecting that the primary topic is in a stopped mirror state after synchronizing all the data; after the primary topic is in the stopped mirror state, transitioning the primary topic to the production state; and after the transitioning of the primary topic, transitioning the corresponding secondary topic to the mirror state, the corresponding secondary topic mirroring data from the primary topic