High Availability

A data distribution policy defines how application data is distributed, or partitioned, across multiple nodes. Partitioning of application data allows large application data sets to be distributed across multiple nodes, typically each running on a separate machine. This provides a simple mechanism for applications to grow horizontally by scaling the amount of machine resources available for storing and processing application data. Data distribution policies are associated with one or more availability zones.

To support partitioning application data, an application fragment must be associated with one or more data distribution policies. A data distribution policy defines these characteristics:

Partitions can be distributed across the available machine resources, either dynamically, a dynamic data distribution policy, or statically, a static data distribution policy.

A dynamic data distribution policy automatically creates dynamic partitions based on the available nodes. There is no requirement to explicitly configure dynamic partitions or to map them to specific nodes. Data in a dynamic data distribution policy is automatically redistributed when a node joins or leaves a cluster without any explicit operator action.

Figure 2, “Dynamic data distribution policy” shows a dynamic data distribution policy that provides round-robin data distribution across three nodes. As data is created it is assigned to the nodes using a round-robin algorithm; Data 1 is created on Node A, Data 2 is created on Node B, Data 3 is created on Node C, and the next data created will be created back on Node A. There are dynamic partitions created to support this data distribution policy, but they are omitted from the diagram since they are implicitly created by the runtime. If additional nodes are added, the round-robin algorithm will include them when distributing data.

A static data distribution policy requires that static partitions be explicitly configured and mapped to nodes. Data is not automatically redistributed when a node joins or leaves a cluster when using a static data distribution policy. Data is only redistributed across the cluster by loading an updated configuration with the partition node mapping changed.

Figure 3, “Static data distribution policy” shows a static data distribution policy that uses a data mapper that maps odd data to a partition named Odd and even data to a partition named Even. As data is created it is created on the active node for the partition, so all Odd data is created on Node A and all Even data is created on Node B. Adding a new node to a static distribution policy will not impact where the data is created, it will continue to be created on the nodes that are hosting the partition to which the data is mapped. Contrast this to the behavior of the dynamic data distribution policy — where new nodes affect the distribution of the data.

The number of copies of data, or replicas, is defined by a data distribution policy. The policy also defines whether replication should be done synchronously or asynchronously, and other runtime tuning and recovery properties.

Data is distributed across partitions at runtime using a data mapper. A data mapper maps application data to a specific partition at runtime, usually based on the content of the application data. Builtin data mappers are provided, and applications can provide their own data mappers to perform application specific data mapping.

Figure 4, “Single data center availability zone” shows a simple example of a single cluster with three nodes deployed in a single data center. There is one availability zone defined which contains all of the nodes in the cluster. The availability zone would use a data distribution policy appropriate for the deployed application, and possibly define a quorum policy that ensures that quorum is always maintained.

Figure 5, “Disaster recovery (DR) availability zone” shows a more complex example of a cluster that spans two data centers to provide disaster recovery for an application. This examples defines these availability zones:

Notice that nodes C and E are in multiple availability zones, with each availability zone having possibly different quorum policies.

The application data types that represent the critical application data must be in the same data distribution policy. This distribution policy is then associated with the Disaster Recovery availability zone to ensure that the data is redundant across data centers.

Data is partitioned by installing a data mapper on a managed object using configuration. A managed object with a data mapper installed is called a partitioned object. A data mapper is responsible for assigning a managed object to a partition. Partition assignment occurs:

Data mappers are inherited by all subtypes of a parent type. A child type can install a new data mapper to override a parent's data mapper.

A partitioned object is always associated with a single partition, but the partition it is associated with can change during the lifetime of the object.

The algorithm used by a data mapper to assign an object to a partition is application specific. It can use any of the following criteria to make a partition assignment:

Builtin data mappers for consistent hashing and round-robin are provided.

Figure 9, “Foreign partition” shows three nodes, A, B, and C, and a partition P. Partition P is defined with an active node of A and a replica node of B. Partition P is also defined on node C but node C is not in the node list for the partition. On node C, partition P is considered a foreign partition.

The partition state and node list of foreign partitions is maintained as the partition definition changes in the cluster. However, no objects are replicated to these nodes, and these nodes cannot become the active node for the partition. When an object in a foreign partition is created or updated, the create and update is pushed to the active and any replica nodes in the partition.

Foreign partition definitions are useful for application specific mechanisms that require a node to have a distributed view of partition state, without being the active node or participating in replication.

Partitions are defined directly by the application or by configuration. Partitions should be defined and enabled (see Enabling and Disabling Partitions) on all nodes on which the partition should be known. This allows an application to:

As an example, here are the steps to define a partition P in a cluster with an active node of A and a replica node of B.

Partition definitions can be redefined to allow partitions to be migrated to different nodes. See Migration for details.

The only time that node list inconsistencies are detected is when data rebalancing is done (see Rebalancing), or a foreign partition (see Foreign Partitions) is being defined.

Once a partition is defined it can never be removed — it can only be redefined. This ensures that data assigned to a partition is never abandoned because a partition is deleted.

Once, a partition has been defined, it must be enabled. Enabling a partition causes the local node to transition the partition from the Initial state to the Active state. Partition activation may include migration of object data from other nodes to the local node. It may also include updating the active node for the partition in the cluster. Enabling an already Active partition has no affect.

Disabling a partition causes the local node to stop hosting the partition. The local node is removed from the node list in the partition definition on all nodes in the cluster. If the local node is the active node for a partition, the partition will migrate to the next node in the node list and become active on that node. As part of migrating the partition all objects in the partition on the local node are removed from shared memory.

When a partition is disabled with only the local node in the node list there is no impact to the objects contained in the partition on the local node since a partition migration does not occur. These objects can continue to be read by the application. However, unless the partition mapper is removed, no new objects can be created in the disabled partition because there is no active node for the partition.

A partition with one or more replica nodes defined in its node list will failover if its current active node fails. The next highest priority available node in the node list will take over processing for this partition.

When a node fails, it is removed from the node list for the partition definition in the cluster. All undiscovered nodes in the node list for the partition are also removed from the partition definition. such as, if node A fails with the partition definitions in Figure 6, “Partition definitions” active, the node list is updated to remove node A leaving these partition definitions active in the cluster.

Once a node has been removed from the node list for a partition, no communication occurs to that node for the partition.

A node is restored to service by loading and activating configuration that defines the data distribution policies, availability zones, and any static partition to node mappings used by the node. When the configuration is activated, other nodes in the cluster are contacted and their partition definitions are updated to reflect the new cluster topology based on the node configuration of the node being restored. When a partition definition changes because of this configuration activation, partition migration occurs, which copies data as required to reflect any new or updated partition definitions. The partition migration will cause any data hosted on the node being restored, either active or replica data, to be copied to the node.

Partitioned data in an availability zone can be rebalanced on an active system without impacting application availability. Rebalancing redistributes application data across all nodes in the availability zone based on the data distribution policies and the current number of nodes in the availability zone.

Rebalancing occurs:

When rebalancing occurs, the data mapper associated with the partitions being rebalanced is called for all data in the partitions. The data mapper may map the data to a different partition causing the data to migrate as needed to other nodes in the availability zone.

The number of replica nodes is maintained during rebalancing even if a node providing support for replica data is no longer part of an availability zone — in this case the replica data is migrated to another node in the availability zone.

Partitions support migration to different nodes without requiring system downtime. Partition migration is initiated as part of restoring nodes to service (see Restore) and rebalancing partitioned data (see Rebalancing). Partition migration supports the following changes to a partition definition:

When partition migration is initiated all data is copied as required to support the updated partition definition, this may include changing the active node for the partition.

such as, these steps will migrate a partition P with an active node of A, and a replica of B, from node A to C:

When the partition migration is complete, partition P is now active on node C, with node B still the replica. Node A is no longer hosting this partition.

It is also possible to force replication to all replica nodes during partition migration by setting the force replication configuration property when loading new configuration. Setting the force replication property will cause all replica nodes to be resynchronized with the active node during partition migration. In general forcing replication is not required since replica nodes resynchronize with the active node when partitions are defined on a replica node.

Partitions can have one of the following states:

Figure 11, “Partition state machine” shows the state machine that controls the transitions between all of these states.

The external events in the state machine map to an API call or an administrator command. The internal events are generated as part of node processing.

Partitions also have a status, which defines how the local definition of the partition was done, and whether it has been enabled. The valid states are defined in Partition status.

All of the partition status values are controlled by an administrative operation, or API, on the local node except for the RemoteEnabled and RemoteDefined statuses. The RemoteEnabled and RemoteDefined statuses occurs when local partition state was not defined and enabled on the local node, it was only updated on a remote node.

If the local node leaves the cluster and is restarted, it must redefine and enable a partition locally before rejoining the cluster to rejoin as a member of the partition. For this reason it is recommended that all nodes perform define and enable for all partitions in which they participate, even if they are a replica node in the partition.

If an I/O error is detected attempting to send creates, updates, or deletes to a replica node, an error is logged on the node initiating the replication and the object modifications for the replica node are discarded and the replica node is removed from the node list for the partition. These errors include:

The replica node will be re-synchronized with the active node when the replica node is restored (see Restore).

The valid node quorum states are defined in Node quorum states.

Figure 16, “Quorum State Machine” shows the state machine that controls the transitions between all of the node quorum states.

There are cases where disabling node quorum is desired. Examples are:

To support these cases, the node quorum mechanism can be disabled using configuration (see the TIBCO StreamBase Administration Guide). When node quorum is disabled, high availability services will never be disabled on a node because of a lack of quorum. With the node quorum mechanism disabled, a node can only be in the Active or Partial node quorum states defined in Node quorum states — it never transitions to the Disabled state. Because of this, it is possible that partitions may have multiple active nodes simultaneously.

This section describes how to restore a cluster following a multi-master scenario. These terms are used to describe the roles played by nodes in restoring after a multi-master scenario:

To recover partitions that were active on multiple nodes, support is provided for merging objects using an application implemented compensation trigger. If a conflict is detected, the compensation trigger is executed on the source node to allow the conflict to be resolved.

The types of conflicts that are detected are:

The application implemented compensation trigger is always executed on the source node. The compensation trigger has access to data from the initiating and source nodes.

Figure 17, “Active cluster” shows an example cluster with a single partition, P, that has node A as the active node and node B as the replica node.

Figure 18, “Failed cluster” shows the same cluster after connectivity is lost between node A and node B with node quorum disabled. The partition P definition on node A has been updated to remove node B as a replica because it is no longer possible to communicate with node B. Node B has removed node A from the partition definition because it believes that node A has failed so it has taken over responsibility for partition P.

Once connectivity has been restored between all nodes in the cluster, and the nodes have discovered each other, the operator can initiate the restore of the cluster. The restore (see Restore) is initiated on the initiating node which is node A in this example. All partitions on the initiating node are merged with the same partitions on the source nodes on which the partitions are also active. In the case where a partition was active on multiple remote nodes, the node to merge from can be specified per partition, when the restore is initiated. If no remote node is specified for a partition, the last remote node to respond to the Is partition(n) active? request (see Figure 19, “Merge operation - using broadcast partition discovery”) will be the source node.

Figure 19, “Merge operation - using broadcast partition discovery” shows the steps taken to restore the nodes in Figure 18, “Failed cluster”. The restore command is executed on node A which is acting as the initiating node. Node B is acting as the source node in this example.

The steps in Figure 19, “Merge operation - using broadcast partition discovery” are:

The steps to restore a node, when the restore from node was specified in the restore operation are very similar to the ones above, except that instead of a broadcast to find the source node, a request is sent directly to the specified source node.

The example in this section has the A node as the final active node for the partition. However, there is no requirement that this is the case. The active node for a partition could be any other node in the cluster after the restore completes, including the source node.

Figure 20, “Split cluster” shows another possible multi-master scenario where the network outage causes a cluster to be split into multiple sub-clusters. In this diagram there are two sub-clusters:

To restore this cluster, the operator must decide which sub-cluster nodes should be treated as the initiating nodes and restore from the source nodes in the other sub-cluster. The steps to restore the individual nodes are identical to the ones described above.