== Clustering
Infinispan can be configured to be either local (standalone) or clustered.
If in a cluster, the cache can be configured to replicate changes to all nodes, to
invalidate changes across nodes and finally to be used in distributed mode - state
changes are replicated to a small subset of nodes enough to be fault tolerant but not
too many nodes to prevent scalability.


=== Local Mode
While Infinispan is particularly interesting in clustered mode, it also offers a very
capable local mode.
In this mode, it acts as a simple, in-memory data cache similar to JBoss Cache and EHCache.

But why would one use a local cache rather than a map? Caches offer a lot of features
over and above a simple map, including write-through and write-behind caching to persist
data, eviction of entries to prevent running out of memory, and support for expirable
entries.
Infinispan, specifically, is built around a high-performance, read-biased data container
which uses modern techniques similar to
link:http://en.wikipedia.org/wiki/Read-copy-update[read-copy-update]
-- which buys you non-blocking, thread-safe reads even when concurrent writes are taking
place.
Infinispan also makes heavy use of compare-and-swap and other lock-free algorithms,
making it ideal for high-throughput, multi-CPU/multi-core environments.
Further, Infinispan's Cache API extends the JDK's ConcurrentMap - making migration from a
map to Infinispan trivial.


=== Replicated Mode
Replication is a simple clustered mode where cache instances automatically discover
neighboring instances on other JVMs on the same local network, and form a cluster.
Entries added to any of these cache instances will be replicated to all other cache
instances in the cluster, and can be retrieved locally from any instance.
This clustered mode provides a quick and easy way to share state across a cluster,
however replication practically only performs well in small clusters (under 10 nodes),
due to the number of replication messages that need to happen - as the cluster size
increases.
Infinispan can be configured to use UDP multicast which mitigates this problem to some
degree.

.Replication mode
image::images/Figure1_6.png[]

Replication can be synchronous or asynchronous. Use of either one of the options is
application dependent.
Synchronous replication blocks the caller (e.g. on a put() ) until the modifications
have been replicated successfully to all nodes in a cluster.
Asynchronous replication performs replication in the background (the put() returns
immediately).
Infinispan offers a replication queue, where modifications are replicated periodically
(i.e. interval-based), or when the queue size exceeds a number of elements, or a
combination thereof.
A replication queue can therefore offer much higher performance as the actual replication
is performed by a background thread.

Asynchronous replication is faster (no caller blocking), because synchronous replication
requires acknowledgments from all nodes in a cluster that they received and applied the
modification successfully (round-trip time).
However, when a synchronous replication returns successfully, the caller knows for sure
that all modifications have been applied to all cache instances, whereas this is not be
the case with asynchronous replication. With asynchronous replication, errors are simply
written to a log.
Even when using transactions, a transaction may succeed but replication may not succeed
on all cache instances.


=== Invalidation Mode
Invalidation is a clustered mode that does not actually share any data at all, but simply
aims to remove data that may be stale from remote caches. This cache mode only makes
sense if you have another, permanent store for your data such as a database and are only
using Infinispan as an optimization in a read-heavy system, to prevent hitting the
database every time you need some state. If a cache is configured for invalidation rather
than replication, every time data is changed in a cache other caches in the cluster
receive a message informing them that their data is now stale and should be evicted from
memory.

.Invalidation mode
image::images/Figure2_5.png[]

Invalidation, when used with a shared cache loader would cause remote caches to refer to
the shared cache loader to retrieve modified data. The benefit of this is twofold:
network traffic is minimized as invalidation messages are very small compared to
replicating updated data, and also that other caches in the cluster look up modified
data in a lazy manner, only when needed.

Invalidation messages are sent after each modification (no transactions or batches), or
at the end of a transaction or batch, upon successful commit.
This is usually more efficient as invalidation messages can be optimized for the
transaction as a whole rather than on a per-modification basis.

Invalidation too can be synchronous or asynchronous, and just as in the case of
replication, synchronous invalidation blocks until all caches in the cluster receive
invalidation messages and have evicted stale data while asynchronous invalidation works
in a 'fire-and-forget' mode, where invalidation messages are broadcast but doesn't block
and wait for responses.


=== Distribution Mode
Distribution is a powerful clustering mode which allows Infinispan to scale linearly as
more nodes are added to the cluster.
Distribution makes use of a hash algorithm to determine on which node(s) entries should
be stored.
The number of copies that should be maintained in the cluster for each cache entry is
configurable (*numOwners*).
The number of copies represents the trade-off between performance and durability of data.
The more copies you maintain, the lower performance will be, but also the lower the risk
of losing data due to server outages.
Regardless of how many copies are maintained, distribution still scales linearly and this
is key to Infinispan's scalability.

Another feature of the hash algorithm is that it is deterministic in locating entries
without resorting to multicast requests or maintaining expensive metadata.
Doing a GET anywhere will result in at most _numOwners_ remote calls.
The remote GET requests are staggered: we request the value from the primary owner,
but if it doesn't respond in a reasonable amount of time, we request the value from
the backup owners as well.
(The {{infinispan.stagger.delay}} system property, in milliseconds, controls the delay between requests.)
A GET may also result in 0 remote calls if the key is present in the local cache.
Doing a PUT can result in more remote calls, depending on the cache configuration (e.g.
whether the cache is transactional).


==== Read consistency
Since GETs are sent to all data owners in parallel and the first returning result is used,
this can lead to data inconsistency when using an _asynchronous_ transport.
If an updating thread modifies the primary data owner, but updates are only sent to backup
nodes asynchronously, a concurrent read from the same thread may read a stale value for a
short period of time until the asynchronous replication completes.

Note that this is _only_ if the transport is _asynchronous_.
If using a _synchronous_ transport this behavior is not exhibited.

.Distribution mode
image::images/Figure3_3.png[]


==== Hashing Algorithms
The hashing algorithm in Infinispan is based on
link:http://en.wikipedia.org/wiki/Consistent_hashing[consistent hashing], and even
though our implementation has diverged a bit, we still use the term *consistent hash*.

Unlike in consistent hashing, we split the key space into fixed *segments*.
The number of segments is configurable (*numSegments*), and it cannot be changed without
restarting the cluster.
The mapping of keys to segments is also fixed -- a key should map to the same segment,
regardless of how the topology of the cluster changes.

Each hash segment is mapped to a list of nodes called *owners*.
The order matters, because the first owner, also known as the *primary owner*, has a
special role in many cache operations (e.g. locking).
The other owners are called *backup owners*.
There is no hard rule on how the segments are mapped to owners, although the hashing
algorithms generally try to balance the number of segments allocated to each node and at
the same time minimize the number of segments that have to move after a node joins or
leaves the cluster.

The hashing algorithm in Infinispan is customizable, and in fact there are five
implementations that ship with Infinispan by default:

org.infinispan.distribution.ch.DefaultConsistentHashFactory::
  The default hashing algorithm.
  It achieves a pretty even distribution, but it has one disadvantage: the mapping of
  segments to nodes depends on the order in which caches joined the cluster, so a key's
  owners are not guaranteed to be the same in all the caches running in a cluster.
org.infinispan.distribution.ch.TopologyAwareConsistentHashFactory::
  Selected automatically when <<ServerHinting, Server Hinting>> is enabled.
  Similar to the default algorithm, but also tries to spread each segment's copies across
  as many sites, racks, or machines as possible.
org.infinispan.distribution.ch.SyncConsistentHashFactory::
  An alternative algorithm, closer to consistent hashing (but still not exactly the same).
  It addresses the weakness of the default algorithm, and always assigns a key to the same
  node in every cache as long as the cluster is symmetric. It does have some weaknesses of
  itself, though: the load distribution is less even, and it also moves more segments than
  necessary on a join or leave.
org.infinispan.distribution.ch.TopologyAwareSyncConsistentHashFactory::
  Similar to _SyncConsistentHashFactory_, but adapted for <<ServerHinting, Server Hinting>>.
org.infinispan.distribution.ch.ReplicatedConsistentHashFactory::
  This algorithm is used internally to implement replicated caches.
  Users should never select this explicitly in a distributed cache.


===== Hashing Configuration
This is how you configure hashing declaratively, via XML:

[source,xml]
----
   <distributed-cache name="distributedCache" owners="2" segments="100" capacity-factor="2" />
----

And this is how you can configure it programmatically, in Java:
[source,java]
----
Configuration c = new ConfigurationBuilder()
   .clustering()
      .cacheMode(CacheMode.DIST_SYNC)
      .hash()
         .numOwners(2)
         .numSegments(100)
   .build();
----

==== Initial cluster size
Infinispan's very dynamic nature in handling toplogy changes (i.e. nodes being added / removed
at runtime) means that, normally, a node doesn't wait for the presence of other nodes before 
starting. While this is very flexible, it might not be suitable for applications which require
a specific number of nodes to join the cluster before caches are started. For this reason,
you can specify how many nodes should have joined the cluster before proceeding with cache 
initialization. To do this, use the _initialClusterSize_ and _initialClusterTimeout_ transport
properties. The declarative XML configuration:

[source,xml]
----
   <transport initial-cluster-size="4" initial-cluster-timeout="30000" />
----

The programmatic Java configuration:
[source,java]
----
GlobalConfiguration global = new GlobalConfigurationBuilder()
   .transport()
       .initialClusterSize(4)
       .initialClusterTimeout(30000)
   .build();
----

The above configuration will wait for _4_ nodes to join the cluster before initialization. If
the initial nodes do not appear within the specified timeout, the cache manager will fail to 
start.

==== L1 Caching
To prevent repeated remote calls when doing multiple GETs, L1 caching can be enabled.
L1 caching places remotely received values in a near cache for a short period of time
(configurable) so repeated lookups would not result in remote calls.
In the above diagram, if L1 was enabled, a subsequent GET for the same key on Server3
would not result in any remote calls.

.L1 caching
image::images/Figure4_4.png[]

L1 caching is not free though.
Enabling it comes at a cost, and this cost is that every time a key is updated, an
invalidation message needs to be multicast to ensure nodes with the entry in L1
invalidates the entry.
L1 caching causes the requesting node to cache the retrieved entry locally and listen for
changes to the key on the wire.
L1-cached entries are given an internal expiry to control memory usage.
Enabling L1 will improve performance for repeated reads of non-local keys, but will
increase memory consumption to some degree.
It offers a nice tradeoff between the "read-mostly" performance of an invalidated data
grid with the scalability of a distributed one.
Is L1 caching right for you?
The correct approach is to benchmark your application with and without L1 enabled and see
what works best for your access pattern.

TIP: Looking for Buddy Replication?  Buddy Replication - from JBoss Cache - does not exist
in Infinispan. 
See this blog article which discusses the reasons why Buddy Replication was not
implemented in Infinispan, and how the same effects can be achieved using Infinispan:
link:http://infinispan.blogspot.com/2009/08/distribution-instead-of-buddy.html[]


[[ServerHinting]]
==== Server Hinting
The motivations behind this feature is to ensure when using distribution, backups are not
picked to reside on the same physical server, rack or data centre.
For obvious reasons it doesn't work with total replication.

===== Configuration
The hints are configured at transport level:

[source,xml]
----
<transport
    cluster = "MyCluster"
    machine = "LinuxServer01"
    rack = "Rack01"
    site = "US-WestCoast" />
----

The following topology hints can be specified:

machine::
  This is probably the most useful, to disambiguate between multiple JVM instances on the
  same node, or even multiple virtual hosts on the same physical host.
rack::
  In larger clusters with nodes occupying more than a single rack, this setting would help
  prevent backups being stored on the same rack.
site::
  To differentiate between nodes in different data centres replicating to each other. Note
  that <<CrossSiteReplication, Cross site replication>> is another alternative for
  clusters that need to span two or more data centres.

All of the above are optional, and if not provided, the distribution algorithms provide no
guarantees that backups will not be stored in instances on the same machine/rack/site.


[[KeyAffinityService]]
====  Key affinity service
The key affinity service solves the following problem: for a distributed Infinispan
cluster one wants to make sure that a value is placed in a certain node.
Based on a supplied cluster
link:http://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/remoting/transport/Address.html[address]
identifying the node, the service returns a key that will be hashed to that particular
node.

===== API
Following code snippet depicts how a reference to this service can be obtained and used.

[source,java]
----
// 1. Obtain a reference to a cache manager
EmbeddedCacheManager cacheManager = getCacheManager();//obtain a reference to a cache manager
Cache cache = cacheManager.getCache();
 
// 2. Create the affinity service
KeyAffinityService keyAffinityService = KeyAffinityServiceFactory.newLocalKeyAffinityService(
      cache,
      new RndKeyGenerator(),
      Executors.newSingleThreadExecutor(),
      100);
 
// 3. Obtain a key to be mapped to a certain address
Object localKey = keyAffinityService.getKeyForAddress(cacheManager.getAddress());
 
// 4. This put makes sure that the key resigns on the local node (as obtained cacheManager.getAddress())
cache.put(localKey, "yourValue");
----

The service is started at step 2: after this point it uses the supplied _Executor_ to
generate and queue keys.
At step 3, we obtain a key for this service, and use it at step 4, with that guarantee
that it is distributed on the node identified by `cacheManager.getAddress()`.

===== Lifecycle
_KeyAffinityService_ extends _Lifecycle_, which allows stopping and (re)starting it:

[source,java]
----
public interface Lifecycle {
   void start();
   void stop();
}
----

The service is instantiated through _KeyAffinityServiceFactory_. All the factory methods
have an _Executor_ parameter, that is used for asynchronous key generation (so that it
won't happen in the caller's thread).
It is the user's responsibility to handle the shutdown of this _Executor_.

The _KeyAffinityService_, once started, needs to be explicitly stopped. This stops the
background key generation and releases other held resources.

The only situation in which _KeyAffinityService_ stops by itself is when the cache manager
with which it was registered is shutdown.

===== Topology changes
When a topology change takes place the key ownership from the _KeyAffinityService_ might
change.
The key affinity service keep tracks of these topology changes and updates and doesn't
return stale keys, i.e. keys that would currently map to a different node than the one
specified.
However, this does not guarantee that at the time the key is used its node affinity hasn't
changed, e.g.:

- Thread `T1` reads a key `k1` that maps to node `A`.
- A topology change happens which makes `k1` map to node `B`.
- `T1` uses `k1` to add something to the cache. At this point `k1` maps to `B`, a different node than the one
  requested at the time of read.

Whilst this is not ideal, it should be a supported behaviour for the application as all
the already in-use keys might be moved over during cluster change.
The _KeyAffinityService_ provides an access proximity optimisation for stable clusters which
doesn't apply during the instability of topology changes.


====  The Grouping API
Complementary to <<KeyAffinityService, Key affinity service>> and similar to
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/atomic/AtomicMap.html[AtomicMap],
the grouping API allows you to co-locate a group of entries on the same nodes, but without
being able to select the actual nodes.

===== How does it work?
Normally, when you store an entry, Infinispan will take a hash code of the key, map the
hash to a hash segment, and store the entry on the nodes which own that segment.
Infinispan always uses an algorithm to locate a key, never allowing the nodes on which the
entry is stored to be specified manually.
This scheme allows any node to know which nodes owns a key, without having to distribute
such ownership information.
This reduces the overhead of Infinispan, but more importantly improves redundancy as there
is no need to replicate the ownership information in case of node failure.

If you use the grouping API, then Infinispan will ignore the hash of the key when deciding
which _node_ to store the entry on, and instead use a hash of the group.
Infinispan still uses the hash of the key in its internal data structures, so using the
grouping API will not slow things down.
When the group API is in use, it is important that every node can still compute, using an
algorithm, the owners of every key.
For this reason, the group cannot be specified manually.
The group can either be intrinsic to the entry (generated by the key class) or extrinsic
(generated by an external function).

===== How do I use the grouping API?
First, you must enable groups. If you are configuring Infinispan programmatically, then call:

[source,java]
----
Configuration c = new ConfigurationBuilder()
   .clustering().hash().groups().enabled()
   .build();
----

Or, if you are using XML:

[source,xml]
----
<distributed-cache>
   <groups enabled="true"/>
</distributed-cache>
----

If you have control of the key class (you can alter the class definition, it's not part of
an unmodifiable library), and the determination of the group is not an orthogonal concern
to the key class, then we recommend you use an intrinsic group.
The intrinsic group can be specified using the _@Group_ annotation placed on the method.
Let's take a look at an example:

[source,java]
----
class User {
   ...
   String office;
   ...

   public int hashCode() {
      // Defines the hash for the key, normally used to determine location
      ...
   }

   // Override the location by specifying a group, all keys in the same
   // group end up with the same owner
   @Group
   public String getOffice() {
      return office;
   }
   }
}
----

NOTE: The group must be a `String`

If you don't have control over the key class, or the determination of the group is an
orthogonal concern to the key class, we recommend you use an extrinsic group.
An extrinsic group is specified by implementing the _Grouper_ interface, which has a
single method computeGroup, which should return the group.
_Grouper_ acts as an interceptor, passing the previously computed value in.
The group passed to the first _Grouper_ will be that determined by _@Group_ (if _@Group_
is defined).
This allows you even greater control over the group when using an intrinsic group.
For a grouper to be used when determining the group for a key, its _keyType_ must be
assignable from the key being grouped.

Let's take a look at an example of a _Grouper_:

[source,java]
----
public class KXGrouper implements Grouper<String> {

   // A pattern that can extract from a "kX" (e.g. k1, k2) style key
   // The pattern requires a String key, of length 2, where the first character is
   // "k" and the second character is a digit. We take that digit, and perform
   // modular arithmetic on it to assign it to group "1" or group "2".
   private static Pattern kPattern = Pattern.compile("(^k)(<a>\\d</a>)$");

    public String computeGroup(String key, String group) {
        Matcher matcher = kPattern.matcher(key);
        if (matcher.matches()) {
            String g = Integer.parseInt(matcher.group(2)) % 2 + "";
            return g;
        } else
            return null;
    }

    public Class<String> getKeyType() {
        return String.class;
    }
}
----

Here we determine a simple grouper that can take the key class and extract from the group
from the key using a pattern.
We ignore any group information specified on the key class.

You must register every grouper you wish to have used. If you are configuring Infinispan
programmatically:

[source,java]
----
Configuration c = new ConfigurationBuilder()
   .clustering().hash().groups().enabled().addGrouper(new KXGrouper())
   .build();
----

Or, if you are using XML:

[source,xml]
----
<distributed-cache>
   <groups enabled="true">
      <grouper class="com.acme.KXGrouper" />
   </groups>
</distributed-cache>
----

===== Advanced Interface

NOTE: This interface is available since Infinispan 7.0.0.

+AdvancedCache+ allows to interact with the keys belonging to a group. It is possible to return the +Set+ of keys
belonging to a group and remove all the keys of the group. Below is the interface available:

[source,java]
.AdvancedCache.java
----
/**
 * It fetches all the keys which belong to the group.
 * <p/>
 * Semantically, it iterates over all the keys in memory and persistence, and performs a read operation in the keys
 * found. Multiple invocations inside a transaction ensures that all the keys previous read are returned and it may
 * return newly added keys to the group from other committed transactions (also known as phantom reads).
 * <p/>
 * The {@code map} returned is immutable and represents the group at the time of the invocation. If you want to add
 * or remove keys from a group use {@link #put(Object, Object)} and {@link #remove(Object)}. To remove all the keys
 * in the group use {@link #removeGroup(String)}.
 * <p/>
 * To improve performance you may use the {@code flag} {@link org.infinispan.context.Flag#SKIP_CACHE_LOAD} to avoid
 * fetching the key/value from persistence. However, you will get an inconsistent snapshot of the group.
 *
 * @param groupName the group name.
 * @return an immutable {@link java.util.Map} with the key/value pairs.
 */
Map<K, V> getGroup(String groupName);

/**
 * It removes all the key which belongs to a group.
 * <p/>
 * Semantically, it fetches the most recent group keys/values and removes them.
 * <p/>
 * Note that, concurrent addition perform by other transactions/threads to the group may not be removed.
 *
 * @param groupName the group name.
 */
void removeGroup(String groupName);
----

===  Asynchronous Options
When Infinispan instances are clustered, regardless of the cluster mode, data can be
propagated to other nodes in a synchronous or asynchronous way.
When synchronous, the sender waits for replies from the receivers and when asynchronous,
the sender sends the data and does not wait for replies from other nodes in the cluster.

With asynchronous modes, speed is more important than consistency and this is particularly
advantageous in use cases such as HTTP session replication with sticky sessions enabled.
In these scenarios, data, or in this case a particular session, is always accessed on the
same cluster node and only in case of failure is data accessed in a different node.
This type of architectures allow consistency to be relaxed in favour of increased
performance.

In order to choose the asynchronous configuration that best suits your application, it's
important to understand the following configuration settings:

==== Asynchronous Communications
Whenever you add
link:http://docs.jboss.org/infinispan/{infinispanversion}/configdocs/infinispan-config-{infinispanversion}.html[`<replicated-cache mode="ASYNC"> or <distributed-cache mode="ASYNC"> or <invalidation-cache mode="ASYNC">`]
element, you're telling the underlying JGroups layer in Infinispan
to use asynchronous communication. What this means is that JGroups will send any
replication/distribution/invalidation request to the wire but will not wait for a reply
from the receiver.

==== Replication Queue
The aim of the replication queue is to batch the individual cache operations and send them
as one, as opposed to sending each cache operation individually.
As a result, replication queue enabled configurations perform generally better compared to
those that have it switched off because less RPC messages are sent, fewer envelopes are
used...etc.
The only real trade off to the replication queue is that the queue is flushed periodically
(based on time or queue size) and hence it might take longer for the
replication/distribution/invalidation to be realised across the cluster.
When replication queue is turned off, data is placed directly on the wire and hence it
takes less for data to arrive to other nodes.

==== Asynchronous API
Finally, the <<_asynchronous_api, Asynchronous API>> can be used to emulate non-blocking
APIs, whereby calls are handed over to a different thread and asynchronous API calls
return to the client immediately.
Using this API can lead to reordering, so you should avoid
calling modifying asynchronous methods on the same keys.

==== Return Values
Regardless of the asynchronous option used, the return values of cache operations are
reliable.
If talking about return values of cache operations that return previous value, the
correctness of these returns are guaranteed as well regardless of the clustering mode.
With replication, the previous value is already available locally, and with distribution,
regardless of whether it's asynchronous or synchronous, Infinispan sends a synchronous
request to get the previous value if not present locally.
If on the other hand the asynchronous API is used, client code needs to get hold of the
link:https://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/commons/util/concurrent/NotifyingFuture.html[NotifiyngFuture]
returned by the async operation in order to be able to query the previous value.


=== Partition handling

An Infinispan cluster is built out of a number of nodes where data is stored. In order
not to lose data in the presence of node failures, Infinispan copies the same data - cache
entry in Infinispan parlance - over multiple nodes. This level of data redundancy is
configured through the *numOwners* configuration attribute and ensures that as long as
fewer than *numOwners* nodes crash simultaneously, Infinispan has a copy of the data
available.

However there might be catastrophic situations in which more than *numOwners* nodes
disappear from the cluster:

- split brain. Caused e.g. by a router crash, this splits the cluster in two or more
partitions, or sub-clusters that operate independently. In these circumstances,
multiple clients reading/writing from different partitions see different versions
of the same cache entry, which for many application is problematic. Note there are
ways to alleviate the possibility for the split brain to happen, such as redundant networks or
 link:https://access.redhat.com/documentation/en-US/Red_Hat_Enterprise_Linux/6/html/Deployment_Guide/s2-networkscripts-interfaces-chan.html[IP bonding].
 These only reduce the window of time for the problem to occur, though.
- *numOwners* nodes crash in sequence. When at least *numOwners* nodes crash in rapid
sequence and Infinispan does not have the time to properly rebalance its state between
crashes, the result is partial data lost.

The partition handling functionality discussed in this section allows the user to
be informed when data has been lost, temporarily or permanently, and wait for the cluster to heal.
The goal is to avoid situations in which wrong data is returned to the user as a result of either split
brain or multiple nodes crashing in rapid sequence.
In terms of Brewer's
link:http://en.wikipedia.org/wiki/CAP_theorem[CAP theorem]
, enabling partition handling in Infinispan preserves data consistency
but sacrifices availability in the presence of partitions.

Enabling partition handling is critical for applications that have high consistency
requirements: when the data read from the system must be accurate.
On the other hand, if Infinispan is used as a best-effort cache, partitions are
perfectly tolerable.

The following sections describe the way Infinispan handles <<split-brain,split brain>> and
<<successive-node-failures,successive failures>> when partition handling is enabled, followed by a section on
<<partition-handling-configuration,configuring the partition handling functionality>>.

[[split-brain]]
==== Split brain

In a split brain situation, each network partition will install its own
JGroups view, removing the nodes from the other partition(s).
We don't have a direct way of determining whether the has been split into
two or more partitions, since the partitions are unaware of each other.
Instead, we assume the cluster has split when one or more nodes
disappear from the JGroups cluster without sending an explicit leave message.

With partition handling disabled, each such partition would continue to
function as an independent cluster. Each partition may only see a part of
the data, and each partition could write conflicting updates in the cache.

With partition handling enabled, if we detect a split, each partition does
not start a rebalance immediately, but first it checks whether it should
enter degraded mode instead:

- If at least one segment has lost all its owners (meaning at least
  _numOwners_ nodes left since the last rebalance ended), the partition enters
  degraded mode.
- If the partition does not contain a simple majority of the nodes (floor(numNodes/2) + 1)
  in the _latest stable topology_, the partition also enters degraded mode.
- Otherwise the partition keeps functioning normally, and it starts a rebalance.

The _stable topology_ is updated every time a rebalance operation ends and the coordinator determines
that another rebalance is not necessary.

These rules ensures that at most one partition stays in available mode, and
the other partitions enter degraded mode.

When a partition is in degraded mode, it only allows access to the keys that are wholly owned:

- Requests (reads and writes) for entries that have all the copies on nodes within
this partition are honoured.
- Requests for entries that are partially or totally owned by nodes that disappeared
are rejected with an `AvailabilityException`.

This guarantees that partitions cannot write different values for the same key
(cache is consistent), and also that one partition can not read keys that have been
updated in the other partitions (no stale data).

To exemplify, consider the initial cluster `M = {A, B, C, D}`, configured in distributed
mode with `numOwners = 2`.
Further on, consider three keys `k1`, `k2` and `k3` (that might exist in the cache or not)
such that `owners(k1) = {A,B}`, `owners(k2) = {B,C}` and `owners(k3) = {C,D}`.
Then the network splits in two partitions, `N1 = {A, B}` and `N2 = {C, D}`, they enter
degraded mode and behave like this:

- on `N1`, `k1` is available for read/write, `k2` (partially owned) and `k3` (not owned)
are not available and accessing them results in an `AvailabilityException`
- on `N2`, `k1` and `k2` are not available for read/write, `k3` is available

A relevant aspect of the partition handling process is the fact that when a
split brain happens, the resulting partitions rely on the original consistent
hash function (the one that existed before the split brain) in order
to calculate key ownership. So it doesn't matter if `k1`, `k2`, or `k3` already
existed cache or not, their availability is the same.

If at a further point in time the network heals and `N1` and `N2` partitions
merge back together into the initial cluster `M`, then `M` exits the degraded
mode and becomes fully available again.

As another example, the cluster could split in two partitions `O1 = {A, B, C}`
and `O2 = {D}`, partition `O1` will stay fully
available (rebalancing cache entries on the remaining members).
Partition `O2`, however, will detect a split and enter the degraded mode.
Since it doesn't have any fully owned keys, it will reject any read or write
operation with an `AvailabilityException`.

If afterwards partitions `O1` and `O2` merge back into `M`, then the cache
entries on `D` will be wiped (since they could be stale).
`D` will be fully available, but it will not hold any data until the cache
is rebalanced.

===== Current limitations
Two partitions could start up isolated, and as long as they don't merge they
can read and write inconsistent data. In the future, we will allow custom
availability strategies (e.g. check that a certain node is part of the cluster,
or check that an external machine is accessible) that could handle that
situation as well.


[[successive-node-failures]]
==== Successive nodes stopped
As mentioned in the previous section, Infinispan can't detect whether a node
left the JGroups view because of a process/machine crash, or because of a
network failure: whenever a node leaves the JGroups cluster abruptly, it is
assumed to be because of a network problem.

If the configured number of copies (_numOwners_) is greater than 1, the
cluster can remain available and will try to make new replicas of the data
on the crashed node. However, other nodes might crash during the rebalance process.
If more than _numOwners_ nodes crash in a short interval of time, there is a
chance that some cache entries have disappeared from the cluster altogether.
In this case, with partition handling functionality enabled, Infinispan
assumes (incorrectly) that there is a split brain and enters degraded mode
as described in the split-brain section.

The administrator can also shut down more than *numOwners* nodes in
rapid succession, causing the loss of the data stored only on those nodes.
When the administrator shuts down a node gracefully, Infinispan knows that
the node can't come back.
However, the cluster doesn't keep track of how each node left, and the cache
still enters _degraded_ mode as if those nodes had crashed.

At this stage there is no way for the cluster to recover its state,
except stopping it and repopulating it on restart with the data from an
external source.
Clusters are expected to be configured with an appropriate *numOwners* in
order to avoid *numOwners* successive node failures, so this situation
should be pretty rare.
If the application can handle losing some of the data in the cache, the
administrator can force the availability mode back to AVAILABLE
<<partition-handling-monitoring,via JMX>>.

[[partition-handling-configuration]]
==== Configuration for partition handling functionality

At this stage the partition handling is disabled by default. We will revisit
this decision in the future, based on user feedback.
In order to enable partition handling within the XML configuration:

[source,xml]
----
<distributed-cache name="the-default-cache">
   <partition-handling enabled="true"/>
</distributed-cache>
----

Unless the cache is distributed or replicated, the configuration is ignored.

The same can be achieved programmatically:
[source,java]
----
ConfigurationBuilder dcc = new ConfigurationBuilder();
dcc.clustering().partitionHandling().enabled(true);
----

[[partition-handling-monitoring]]
==== Monitoring and administration

The availability mode of a cache is exposed in JMX as an attribute in the
link:http://docs.jboss.org/infinispan/7.0/apidocs/jmxComponents.html#Cache[Cache MBean].
The attribute is writable, allowing an administrator to forcefully migrate
a cache from degraded state back to available (at the cost of
consistency).

The availability mode is also accessible via the
link:http://docs.jboss.org/infinispan/{infinispanversion}/apidocs/org/infinispan/AdvancedCache.html[AdvancedCache]
interface:

[source,java]
----
AdvancedCache ac = cache.getAdvancedCache();

// Read the availability
boolean available = ac.getAvailability() == AvailabilityMode.AVAILABLE;

// Change the availability
if (!available) {
   ac.setAvailability(AvailabilityMode.AVAILABLE);
}
----


