ZooKeeper Discovery

❗️

This is a legacy Apache Ignite documentation

The new documentation is hosted here: https://ignite.apache.org/docs/latest/

Overview

Apache Ignite default TCP/IP Discovery organizes cluster nodes into a ring-topology form that has its advantages and disadvantages. For instance, on topologies with hundreds of cluster nodes, it can take many seconds for a system message to traverse through all the nodes. As a result, the basic processing of events such as joining of new nodes or detecting the failed ones can take a while affecting the overall cluster responsiveness and performance.

ZooKeeper Discovery is designed for massive Ignite deployments that need to preserve ease of scalability and linear performance. However, using both Ignite and ZooKeeper requires configuring and managing two distributed systems, which can be challenging. Therefore, it is recommended that you use this Discovery SPI only if you plan to scale to 100s and 1000s nodes. Otherwise, it is best to use TCP/IP Discovery.

ZooKeeper Discovery uses ZooKeeper as a single point of synchronization and to organize an Ignite cluster into a star-shaped topology where a ZooKeeper cluster sits in the center and Ignite nodes exchange discovery events through it.

It is worth mentioning that ZooKeeper Discovery is merely an alternate implementation of Ignite Discovery SPI that doesn't affect Ignite Communication SPI (see network configuration). Once the nodes discover each other via ZooKeeper Discovery, they will utilize Communication SPI for peer-to-peer communication.

Configuration

To enable ZooKeeper Discovery, you need to configure ZookeeperDiscoverySpi in a way similar to this one:

<bean class="org.apache.ignite.configuration.IgniteConfiguration">
  ...
  <property name="discoverySpi">
    <bean class="org.apache.ignite.spi.discovery.zk.ZookeeperDiscoverySpi">
      <property name="zkConnectionString" value="127.0.0.1:34076,127.0.0.1:43310,127.0.0.1:36745"/>
      <property name="sessionTimeout" value="30000"/>
      <property name="zkRootPath" value="/apacheIgnite"/>
      <property name="joinTimeout" value="10000"/>
    </bean>
  </property>
</bean>

ZookeeperDiscoverySpi zkDiscoSpi = new ZookeeperDiscoverySpi();

zkDiscoSpi.setZkConnectionString(
  "127.0.0.1:34076,127.0.0.1:43310,127.0.0.1:36745");
zkDiscoSpi.setSessionTimeout(30_000);

zkDiscoSpi.setZkRootPath("");
zkDiscoSpi.setJoinTimeout(10_000);

IgniteConfiguration cfg = new IgniteConfiguration();

//Override default discovery SPI.
cfg.setDiscoverySpi(zkDiscoSpi);

// Start Ignite node.
Ignition.start(cfg);

The following parameters are required (other parameters are optional):

• zkConnectionString - keeps the list of addresses of ZooKeeper servers.
• sessionTimeout - implies the time after which an Ignite node will be considered disconnected if it doesn't react to events exchanged via Discovery SPI.​

Failures and Split Brain Handling

In the situation of network partitioning, some of ​the nodes won't be able to communicate to each other because they are located in separated network segments, which may lead to failure to process user requests or inconsistent data modification.

ZooKeeper Discovery approaches network partitioning (aka. split brain) and communication failures between individual nodes in the following way:

📘

It is assumed that the ZooKeeper cluster is always visible to all the nodes in the cluster. In fact, if a node disconnects from ZooKeeper, it shuts down and other nodes treat it as failed or disconnected.

Whenever a node discovers that it cannot connect to some of the other nodes in the cluster, it initiates a communication failure resolve process by publishing special requests to the ZooKeeper cluster. When the process is started, all nodes try to connect to each other and send the results of the connection attempts to the node that coordinates the process (the coordinator node). Based on this information, the coordinator node creates a connectivity graph that represents the network situation in the cluster. Further actions depend on the type of network segmentation.

The cluster is split into several disjoint components

If the cluster is split into several independent components, each component (being a cluster) may think of itself as a master cluster and continue to process user requests, resulting in data inconsistency. To avoid this, only the component with the largest number of nodes is kept alive; and the nodes from the other components are brought down.

The image above shows a case where the cluster network is split into 2 segments. The nodes from the smaller cluster (right-hand segment) will be terminated.

When there are multiple largest components, the one that has the largest number of clients is kept alive, and the others are shut down.

Several links between nodes are missing

Some nodes cannot connect to some other nodes, which means the nodes are not completely disconnected from the cluster but can't exchange data with some of the nodes and, therefore, cannot be part of the cluster. In the image below, one node cannot connect to two other nodes.

In this case, the task is to find the largest component in which every node can connect to every other node, which, in the general case, is a difficult problem and cannot be solved in acceptable time. The coordinator node uses a heuristic algorithm to find the best approximate solution. The nodes that are left out of the solution are shut down.

ZooKeeper cluster segmentation

In large-scale deployments where the ZooKeeper cluster can span multiple data centers and geographically diverse locations, it can split into multiple segments due to network segmentation. If this occurs, ZooKeeper will check if there is a segment that contains more than a half of all ZooKeeper nodes (this many is required for ZooKeeper to continue its operation), and, if found, this segment will take over managing the Ignite cluster, while other segments will be shut down. If there is no such segment, ZooKeeper will shut down all its nodes.

In case of ZooKeeper cluster segmentation, the Ignite cluster may or may not be split. In any case, when the ZooKeeper nodes are shut down, the corresponding Ignite nodes will try to connect to available ZooKeeper nodes and will shut down if unable to do so.

The following image is an example of network segmentation that splits both the Ignite cluster and ZooKeeper cluster into two segments. This may happen if your clusters are deployed in two data centers. In this case, the ZooKeeper node located in Data Center B will shut itself down. The Ignite nodes located in Data Center B will not be able to connect to the remaining ZooKeeper nodes and will also shut themselves down.

Custom Discovery Events

Changing a ring-shaped topology to the star-shaped one affected the way custom discovery events are handled by Discovery SPI component. Since the ring topology is linear, it means that each discovery message is processed by nodes sequentially. So at any given moment in time, there will be only one node handling a message.

With ZooKeeper Discovery, the coordinator sends discovery messages to all nodes simultaneously resulting in the messages to be processed in parallel.

As a result of this parallel processing, ZooKeeper Discovery doesn't allow to mutate custom discovery events. For instance, the nodes are not allowed to add any payload to discovery messages.

Ignite and ZooKeeper Configuration Considerations

By using ZooKeeper Discovery, you need to be sure that configuration parameters of both systems match each other and don't contradict.

Consider a sample ZooKeeper configuration, as follows:

# The number of milliseconds of each tick
tickTime=2000

# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5

Configured this way, ZooKeeper server detects its own segmentation from the rest of ZooKeeper cluster only after tickTime * syncLimit elapses. Until this event is detected at ZooKeeper level, all Ignite nodes connected to the segmented ZooKeeper server won't try to reconnect to the other ZooKeeper servers.

On the other hand, there is a sessionTimeout parameter on Ignite side that defines how soon ZooKeeper closes an Ignite node's session if the node gets disconnected from the ZooKeeper cluster. If sessionTimeout is smaller than tickTime * syncLimit, then the Ignite node will be notified by the segmented ZooKeeper server too late - its session will expire before it tries to reconnect to other ZooKeeper servers.

To avoid this situation, sessionTimeout should be bigger than tickTime * syncLimit.