Reliable group communication with JGroups

To use log4j2, the API and CORE JARs have to be found on the classpath. There’s an XML configuration for log4j2 in the conf dir, which can be used e.g. via -Dlog4j.configurationFile=$JGROUPS/conf/log4j2.xml.

log4j2 is currently the preferred logging library used by JGroups, and will be used even if the log4j JAR is also present on the classpath.

To use log4j, the log4j JAR has to be found on the classpath. Note though that if the log4j2 API and CORE JARs are found, then log4j2 will be used, so those JARs will have to be removed if log4j is to be used. There’s an XML configuration for log4j in the conf dir, which can be used e.g. via -Dlog4j.configuration=file:$JGROUPS/conf/log4j.properties.

To force use of JDK logging, even if the log4j(2) JARs are present, -Djgroups.use.jdk_logger=true can be used.

JGroups allows custom loggers to be used instead of the ones supported by default. To do this, interface CustomLogFactory has to be implemented:

The implementation needs to return an implementation of org.jgroups.logging.Log.

To use the custom log, LogFactory.setCustomLogFactory(CustomLogFactory f) needs to be called.

It is possible to set the preferred log class via system property jgroups.log_class. To do this, the fully qualified name of a class which provides the following functionality has to be given:

Example: -Djgroups.log_class=org.jgroups.logging.Slf4jLogImpl

The first method takes an object as argument and serializes it into a byte buffer (the object has to be serializable or externalizable). The byte array is then returned. This method is often used to serialize objects into the byte buffer of a message. The second method returns a reconstructed object from a buffer. Both methods throw an exception if the object cannot be serialized or unserialized.

The first method takes an object and writes it to an output stream. The second method takes an input stream and reads an object from it. Both methods throw an exception if the object cannot be serialized or unserialized.

The MessageListener interface below provides callbacks for message reception and for providing and setting the state:

Method receive() is called whenever a message is received and receive(MessageBatch) is called when a message batch is received.

The getState() and setState() methods are used to fetch and set the group state (e.g. when joining). Refer to State transfer for a discussion of state transfer.

The MembershipListener interface is similar to the MessageListener interface above: every time a new view, a suspicion message, or a block event is received, the corresponding method of the class implementing MembershipListener will be called.

Oftentimes the only callback that needs to be implemented will be viewAccepted() which notifies the receiver that a new member has joined the group or that an existing member has left or crashed. The suspect() callback is invoked by JGroups whenever a member if suspected of having crashed, but not yet excluded ^[1].

The block() method is called to notify the member that it will soon be blocked sending messages. This is done by the FLUSH protocol, for example to ensure that nobody is sending messages while a state transfer or view installation is in progress. When block() returns, any thread sending messages will be blocked, until FLUSH unblocks the thread again, e.g. after the state has been transferred successfully.

Therefore, block() can be used to send pending messages or complete some other work. Note that block() should be brief, or else the entire FLUSH protocol is blocked.

The unblock() method is called to notify the member that the FLUSH protocol has completed and the member can resume sending messages. If the member did not stop sending messages on block(), FLUSH simply blocked them and will resume, so no action is required from a member. Implementation of the unblock() callback is optional.

A Receiver can be used to receive messages and view changes; receive() will be invoked as soon as a message has been received, and viewAccepted() will be called whenever a new view is installed.

This class implements Receiver with no-op implementations. When implementing a callback, we can simply extend ReceiverAdapter and overwrite receive() in order to not having to implement all callbacks of the interface.

ReceiverAdapter looks as follows:

A ReceiverAdapter is the recommended way to implement callbacks.

A class implementing ChannelListener can use the JChannel.addChannelListener() method to register with a channel to obtain information about state changes in a channel. Whenever a channel is closed, disconnected or opened, the corresponding callback will be invoked.

If TP.use_ip_addrs is enabled, then instead of using UUIDs, IpAddressUUIDs are used. These are IpAddresses with some randomness added, to prevent reincarnation (restarting of a member under the same address and port, and therefore having the same identity as the previous member).

The advantage of IpAddressUUIDs is that they always carry their physical address with them, therefore the discovery phase needs to exchange less information and no lookup is needed to translate between UUIDs and IpAddresses.

The downside is that IpAddressUUIDs need more memory. See https://issues.jboss.org/browse/JGRP-2080 for details.

The ViewId is used to uniquely number views. It consists of the address of the view creator and a sequence number. ViewIds can be compared for equality and put in a hashmaps as they implement equals() and hashCode().

Whenever a group splits into subgroups, e.g. due to a network partition, and later the subgroups merge back together, a MergeView instead of a View will be received by the application. MergeView is a subclass of View and contains as additional instance variables the list of views that were merged.

As an example if the cluster with view V1={P,Q,R,S,T} split into subgroups V2={P,Q,R} and V2={S,T}, the merged view might be V3={P,Q,R,S,T}. In this case the MergeView contains a list of two views: V2={P,Q,R}) and V2={S,T}.

A channel is created using one of its public constructors (e.g. new JChannel()).

The most frequently used constructor of JChannel looks as follows:

The props argument points to an XML file containing the configuration of the protocol stack to be used. This can be a String, but there are also other constructors which take for example a DOM element or a URL (see the javadoc for details).

The code sample below shows how to create a channel based on an XML configuration file:

If the props argument is null, the default properties will be used. An exception will be thrown if the channel cannot be created. Possible causes include protocols that were specified in the property argument, but were not found, or wrong parameters to protocols.

For example, the Draw demo can be launched as follows:

or

In the latter case, an application downloads its protocol stack specification from a server, which allows for central administration of application properties.

A sample XML configuration looks like this (edited from udp.xml):

A stack is wrapped by <config> and </config> elements and lists all protocols from bottom (UDP) to top (FRAG2). Each element defines one protocol.

Each protocol is implemented as a Java class. When a protocol stack is created based on the above XML configuration, the first element ("UDP") becomes the bottom-most layer, the second one will be placed on the first, etc: the stack is created from the bottom to the top.

Each element has to be the name of a Java class that resides in the org.jgroups.protocols package. Note that only the base name has to be given, not the fully specified class name (UDP instead of org.jgroups.protocols.UDP). If the protocol class is not found, JGroups assumes that the name given is a fully qualified classname and will therefore try to instantiate that class. If this does not work an exception is thrown. This allows for protocol classes to reside in different packages altogether, e.g. a valid protocol name could be com.sun.eng.protocols.reliable.UCAST.

Each layer may have zero or more arguments, which are specified as a list of name/value pairs in parentheses directly after the protocol name. In the example above, UDP is configured with some options, one of them being the IP multicast port (mcast_port) which is set to 45588, or to the value of the system property jgroups.udp.mcast_port, if set.

A channel can be given a logical name which is then used instead of the channel’s address in toString(). A logical name might show the function of a channel, e.g. "HostA-HTTP-Cluster", which is more legible than a UUID 3c7e52ea-4087-1859-e0a9-77a0d2f69f29.

For example, when we have 3 channels, using logical names we might see a view {A,B,C}, which is nicer than {56f3f99e-2fc0-8282-9eb0-866f542ae437,ee0be4af-0b45-8ed6-3f6e-92548bfa5cde, 9241a071-10ce-a931-f675-ff2e3240e1ad}!

If no logical name is set, JGroups generates one, using the hostname and a random number, e.g. linux-3442. If this is not desired and the UUIDs should be shown, use system property -Djgroups.print_uuids=true.

The logical name can be set using:

This must be done before connecting a channel. Note that the logical name stays with a channel until the channel is destroyed, whereas a UUID is created on each connection.

When JGroups starts, it prints the logical name and the associated physical address(es):

The logical name is mac-53465 and the physical address is 192.168.1.3:49932. The UUID is not shown here.

Since 2.12 address generation is pluggable. This means that an application can determine what kind of addresses it uses. The default address type is UUID, and since some protocols use UUID, it is recommended to provide custom classes as subclasses of UUID.

This can be used to for example pass additional data around with an address, for example information about the location of the node to which the address is assigned. Note that methods equals(), hashCode() and compare() of the UUID super class should not be changed.

To use custom addresses, an implementation of org.jgroups.stack.AddressGenerator has to be written.

For any class CustomAddress, it will need to get registered with the ClassConfigurator in order to marshal it correctly:

Set the address generator in JChannel.setAddressGenerator(AddressGenerator). This has to be done before the channel is connected.

An example of a subclass is org.jgroups.util.PayloadUUID, and there are two more shipped with JGroups.

When a client wants to join a cluster, it connects to a channel giving the name of the cluster to be joined:

The cluster name is the name of the cluster to be joined. All channels that call connect() with the same name form a cluster. Messages sent on any channel in the cluster will be received by all members (including the one who sent it).

The connect() method returns as soon as the cluster has been joined successfully. If the channel is in the closed state (see channel states), an exception will be thrown. If there are no other members, i.e. no other member has connected to a cluster with this name, then a new cluster is created and the member joins it as first member. The first member of a cluster becomes its coordinator. A coordinator is in charge of installing new views whenever the membership changes

Clients can also join a cluster and fetch cluster state in one operation. The best way to conceptualize the connect and fetch state connect method is to think of it as an invocation of the regular connect() and getState() methods executed in succession. However, there are several advantages of using the connect and fetch state connect method over the regular connect. First of all, the underlying message exchange is heavily optimized, especially if the flush protocol is used. But more importantly, from a client’s perspective, the connect() and fetch state operations become one atomic operation.

Just as in a regular connect(), the cluster name represents a cluster to be joined. The target parameter indicates a cluster member to fetch the state from. A null target indicates that the state should be fetched from the cluster coordinator. If the state should be fetched from a particular member other than the coordinator, clients can simply provide the address of that member. The timeout paremeter bounds the entire join and fetch operation. An exception will be thrown if the timeout is exceeded.

Method getAddress() returns the address of the channel. The address may or may not be available when a channel is in the unconnected state.

Method getClusterName() returns the name of the cluster which the member joined.

Again, the result is undefined if the channel is in the disconnected or closed state.

The following method can be used to get the current view of a channel:

This method returns the current view of the channel. It is updated every time a new view is installed (viewAccepted() callback).

Calling this method on an unconnected or closed channel is implementation defined. A channel may return null, or it may return the last view it knew of.

Once the channel is connected, messages can be sent using one of the send() methods:

The first send() method has only one argument, which is the message to be sent. The message’s destination should either be the address of the receiver (unicast) or null (multicast). When the destination is null, the message will be sent to all members of the cluster (including itself).

The remainaing send() methods are helper methods; they take either a byte[] buffer or an object, create a Message and call send(Message).

If the channel is not connected, or was closed, an exception will be thrown upon attempting to send a message.

Here’s an example of sending a message to all members of a cluster:

The null value as destination address means that the message will be sent to all members in the cluster. The payload is a hashmap, which will be serialized into the message’s buffer and unserialized at the receiver. Alternatively, any other means of generating a byte buffer and setting the message’s buffer to it (e.g. using Message.setBuffer()) also works.

Here’s an example of sending a unicast message to the first member (coordinator) of a group:

The sample code determines the coordinator (first member of the view) and sends it a "hello world" message.

Methods receive(Message) and receive(MessageBatch) in ReceiverAdapter (or Receiver) can be overridden to receive messages.

A Receiver can be registered with a channel using JChannel.setReceiver(). All received messages, view changes and state transfer requests will invoke callbacks on the registered Receiver:

To receive message batches (see MessageBatch), method receive(MessageBatch) has to be implemented, e.g.:

Implementing the receive(MessageBatch) callback is not strictly necessary, as the default implementation will call receive(Message) for each message of a batch, but it may be more efficient if the application can process batches of messages in one go.

As shown above, the viewAccepted() callback of ReceiverAdapter can be used to get callbacks whenever a cluster membership change occurs. The receiver needs to be set via JChannel.setReceiver(Receiver).

As discussed in ReceiverAdapter, code in callbacks must avoid anything that takes a lot of time, or blocks; JGroups invokes this callback as part of the view installation, and if this user code blocks, the view installation would block, too.

A newly joined member may want to retrieve the state of the cluster before starting work. This is done with getState():

This method returns the state of one member (usually of the oldest member, the coordinator). The target parameter can usually be null, to ask the current coordinator for the state. If a timeout (ms) elapses before the state is fetched, an exception will be thrown. A timeout of 0 waits until the entire state has been transferred.

To participate in state transfer, both state provider and state requester have to implement the following callbacks from ReceiverAdapter (Receiver):

Method getState() is invoked on the state provider (usually the coordinator). It needs to write its state to the output stream given. Note that output doesn’t need to be closed when done (or when an exception is thrown); this is done by JGroups.

The setState() method is invoked on the state requester; this is the member which called JChannel.getState(). It needs to read its state from the input stream and set its internal state to it. Note that input doesn’t need to be closed when done (or when an exception is thrown); this is done by JGroups.

In a cluster consisting of A, B and C, with D joining the cluster and calling Channel.getState(), the following sequence of callbacks happens:

The following code fragment shows how a group member participates in state transfers:

This code is the Chat example from the JGroups tutorial and the state here is a list of strings.

The getState() implementation synchronized on the state (so no incoming messages can modify it during the state transfer), and uses the JGroups utility method objectToStream().

The setState() implementation also uses the Util.objectFromStream() utility method to read the state from the input stream and assign it to its internal list.

Disconnecting from a channel is done using the following method:

It will have no effect if the channel is already in the disconnected or closed state. If connected, it will leave the cluster. This is done (transparently for a channel user) by sending a leave request to the current coordinator. The latter will subsequently remove the leaving node from the view and install a new view in all remaining members.

After a successful disconnect, the channel will be in the unconnected state, and may subsequently be reconnected.

To destroy a channel instance (destroy the associated protocol stack, and release all resources), method close() is used:

Closing a connected channel disconnects the channel first.

The close() method moves the channel to the closed state, in which no further operations are allowed (most throw an exception when invoked on a closed channel). In this state, a channel instance is not considered used any longer by an application and — when the reference to the instance is reset —  the channel essentially only lingers around until it is garbage collected by the Java runtime system.

Every message sending in MessageDispatcher or request invocation in RpcDispatcher is governed by an instance of RequestOptions. This is a class which can be passed to a call to define the various options related to the call, e.g. a timeout, whether the call should block or not, the flags (see Tagging messages with flags) etc.

The various options are:

An example of how to use RequestOptions is:

The methods to send requests are:

castMessage() sends a message to all members defined in dests. If dests is null, the message is sent to all members of the current cluster. If a message is sent synchronously (defined by options.mode), then options.timeout defines the maximum amount of time (in milliseconds) to wait for the responses.

castMessage() returns a RspList, which contains a map of addresses and Rsps; there’s one Rsp per member listed in dests.

A Rsp instance contains the response value (or null), an exception if the target handle() method threw an exception, whether the target member was suspected, or not, and so on. See the example below for more details.

castMessageWithFuture() returns immediately, with a CompletableFuture. The future can be used to fetch the response list (now or later), and it also allows for installation of a callback which will be invoked when the future is done. See Asynchronous calls with futures for details on how to use CompletableFutures.

Note that Buffer is a simple wrapper around a byte[] array, an offset and a length.

sendMessage() allows an application programmer to send a unicast message to a single cluster member and receive the response. The destination of the message has to be non-null (valid address of a member). The mode argument is ignored (it is by default set to ResponseMode.GET_FIRST) unless it is set to GET_NONE in which case the request becomes asynchronous, ie. we will not wait for the response.

sendMessageWithFuture() returns immediately with a future, which can be used to fetch the result.

One advantage of using this building block is that failed members are removed from the set of expected responses. For example, when sending a message to 10 members and waiting for all responses, and 2 members crash before being able to send a response, the call will return with 8 valid responses and 2 marked as failed. The return value of castMessage() is a RspList which contains all responses (not all methods shown):

isReceived() checks whether a response from sender has already been received. Note that this is only true as long as no response has yet been received, and the member has not been marked as failed. numSuspectedMembers() returns the number of members that failed (e.g. crashed) during the wait for responses. getResults() returns a list of return values. get() returns the return value for a specific member.

When a non-null list of addresses is passed (as the destination list) to MessageDispatcher.castMessage() or RpcDispatcher.callRemoteMethods(), then this does not mean that only the members included in the list will receive the message, but rather it means that we’ll only wait for responses from those members, if the call is blocking.

If we want to restrict the reception of a message to the destination members, there are a few ways to do this:

This section shows an example of how to use a MessageDispatcher.

The example starts with the creation of a channel. Next, an instance of MessageDispatcher is created on top of the channel. Then the channel is connected. The MessageDispatcher will from now on send requests, receive matching responses (client role) and receive requests and send responses (server role).

We then send 10 messages to the group and wait for all responses. The timeout argument is 0, which causes the call to block until all responses have been received.

The handle() method simply prints out a message and returns a string. This will be sent back to the caller as a response value (in Rsp.value). Had the call thrown an exception, Rsp.exception would be set instead.

Finally both the MessageDispatcher and channel are closed.

Using reflection to find and invoke methods is rather slow.

As an alternative, we can use method IDs and the MethodLookup or MethodInvoker interfaces to resolve methods, which is faster and has every RPC carry less data across the wire.

Interface MethodLookup looks as follows:

An implementation is given an ID and needs to return the associated Method object. Implementations typically maintain the ID-method mappings in a hashmap and use the method ID as key into the map. This hashmap lookup is faster than having to use Java reflection to find the method for every invocation.

A example of how to use a MethodLookup implementation is shown in RpcDispatcherSpeedTest.

A MethodLookup still uses reflection to invoke the method against the target object. In some cases (e.g. Quarkus), reflection is forbidden, or at least all methods to be invoked via reflection have to be listed at compile-time (when generating the native image). This is tedious when adding/removing fields/methods, and so a way of invoking methods completely free of Java reflection has been added to RpcDispatcher:

An implementation can be set in the RpcDispatcher using setMethodInvoker(MethodInvoker mi). When a MethodInvoker is present in an RpcDispatcher, it takes precedence over MethodLookup.

A MethodInvoker is given the target object, against which to invoke the method, the ID of the method to invoke and a list of arguments. A typical implementation might do the following (copied from ProgrammaticUPerf):

The downside here is that this code needs to be changed when methods are added or removed, or when signatures change. However, if Java reflection cannot be used, then this may be feasible.

The code below shows an example of using RpcDispatcher:

Class RpcDispatcher defines method print() which will be called subsequently. The entry point start() creates a channel and an RpcDispatcher which is layered on top. Method callRemoteMethods() then invokes the remote print() in all cluster members (also in the caller). When all responses have been received, the call returns and the responses are printed.

As can be seen, the RpcDispatcher building block reduces the amount of code that needs to be written to implement RPC-based group communication applications by providing a higher abstraction level between the application and the primitive channels.

Response filters allow application code to hook into the reception of responses from cluster members and can let the request-response execution and correlation code know (1) wether a response is acceptable and (2) whether more responses are needed, or whether the call (if blocking) can return. The RspFilter interface looks as follows:

isAcceptable() is given a response value and the address of the member which sent the response, and needs to decide whether the response is valid (should return true) or not (should return false).

needMoreResponses() determine whether a call returns or not.

The sample code below shows how to use a RspFilter:

Here, we invoke a cluster wide RPC (dests=null), which blocks (mode=GET_ALL) for 10 seconds max (timeout=10000), but also passes an instance of RspFilter to the call (in options).

The filter accepts all responses whose value is greater than 1, and returns as soon as it has received 2 responses which satisfy the above condition.

The following scenario is susceptible to network partitioning and subsequent merging: we have a cluster view of {A,B,C,D} and then the cluster splits into {A,B} and {C,D}. Assume that B and D now acquire a lock "mylock". This is what happens (with the locking protocol being CENTRAL_LOCK):

Therefore the recommended solution here is for nodes to listen to MergeView changes if they expect merging to occur, and re-acquire all of their locks after a merge, e.g.:

With CENTRAL_LOCK2, merging of partitions is handled differently. Contrary to CENTRAL_LOCK, which has the coordinator back up its lock tables to one or more backup members, CENTRAL_LOCK2 doesn’t do this.

Instead, when the current coordinator leaves or crashes, the new coordinator fetches information about locks and pending lock/unlock requests from all members, and then builds its lock table based on this information.

In the above scenario with both B and D holding mylock, in case of a merge (say A becomes the new coordinator), D will be told that its lock mylock has been revoked. This means that D needs to force-unlock D. This can be done in the lockRevoked() callback, e.g.:

This is maginally better than CENTRAL_LOCK, but admittedly less than ideal. Given the following code:

When mylock is revoked, longRunningAction() should be stopped immediately, or - even better - its changes should be undone (like in a transaction). However, this isn’t feasible and would unnecessarily complicate the code.

Here, we see that the Lock abstraction, as easy as it is and as often it is used locally (inside the same JVM), may not be the best abstraction for a distributed setting!

The design of COUNTER is described in detail in CounterService.txt.

In a nutshell, in a cluster the current coordinator maintains a hashmap of named counters. Members send requests (increment, decrement etc) to it, and the coordinator atomically applies the requests and sends back responses.

The advantage of this centralized approach is that - regardless of the size of a cluster - every request has a constant execution cost, namely a network round trip.

A crash or leaving of the coordinator is handled as follows. The coordinator maintains a version for every counter value. Whenever the counter value is changed, the version is incremented. For every request that modifies a counter, both the counter value and the version are returned to the requester. The requester caches all counter values and associated versions in its own local cache.

When the coordinator leaves or crashes, the next-in-line member becomes the new coordinator. It then starts a reconciliation phase, and discards all requests until the reconciliation phase has completed. The reconciliation phase solicits all members for their cached values and versions. To reduce traffic, the request also carries all version numbers with it.

The clients return values whose versions are higher than the ones shipped by the new coordinator. The new coordinator waits for responses from all members or timeout milliseconds. Then it updates its own hashmap with values whose versions are higher than its own. Finally, it stops discarding requests and sends a resend message to all clients in order to resend any requests that might be pending.

There’s another edge case that also needs to be covered: if a client P updates a counter, and both P and the coordinator crash, then the update is lost. To reduce the chances of this happening, COUNTER can be enabled to replicate all counter changes to one or more backup coordinators. The num_backups property defines the number of such backups. Whenever a counter was changed in the current coordinator, it also updates the backups (asynchronously). 0 disables this.

Message bundling is beneficial when sending many small messages; they are queued until a threshold (number of bytes) has been exceeded. Then, the queued messages are assembled into a message batch (see MessageBatch) and the batch is sent.

At the receiver, the message batch is passed up the stack, so protocols and/or the application can process multiple messages in one shot.

When sending many smaller messages, the ratio between payload and message headers might be small; say we send a "hello" string: the payload here is 7 bytes, whereas the addresses and headers (depending on the stack configuration) might be 30 bytes. However, if we bundle (say) 100 messages, then the payload of the large message is 700 bytes, but the header is still 30 bytes. Thus, we’re able to send more actual data across the wire with a message batch than with many small messages.

Message bundling/batching is conceptually similar to TCP’s Nagle algorithm.

A sample configuration is shown below:

Here, bundling is enabled (the default). The max accumulated size is 64'000 bytes.

If at time T0, we’re sending 10 smaller messages with an accumulated size of 2'000 bytes, but then send no more messages, then a message batch of 10 will be sent immediately after the 10th message has been sent.

If we send 1000 messages of 100 bytes each, then - after exceeding 64'000 bytes (after ca. 64 messages) - we’ll send the message batch, and this might have taken only 3 ms.

UDP uses IP multicasting for sending messages to all members of a cluster, and UDP datagrams for unicast messages (sent to a single member). When started, it opens a unicast and multicast socket: the unicast socket is used to send/receive unicast messages, while the multicast socket sends/receives multicast messages. The physical address of the channel will be the address and port number of the unicast socket.

TCP is a replacement for UDP as transport in cases where IP multicast cannot be used. This may be the case when operating over a WAN, where routers might discard IP multicast packets. Usually, UDP is used as transport in LANs, while TCP is used for clusters spanning WANs.

The properties for a typical stack based on TCP might look like this (edited for brevity):

When using TCP, each message to all of the cluster members is sent as multiple unicast messages (one to each member). Due to the fact that IP multicasting cannot be used to discover the initial members, another mechanism has to be used to find the initial membership. There are a number of alternatives (see Initial membership discovery for a discussion of all discovery protocols):

This is a TCP/IP based implementation based on non blocking IO (NIO2).

Details at TCP_NIO2.

Firewalls are usually placed at the connection to the internet. They shield local networks from outside attacks by screening incoming traffic and rejecting connection attempts to host inside the firewalls by outside machines. Most firewall systems allow hosts inside the firewall to connect to hosts outside it (outgoing traffic), however, incoming traffic is most often disabled entirely.

Tunnels are host protocols which encapsulate other protocols by multiplexing them at one end and demultiplexing them at the other end. Any protocol can be tunneled by a tunnel protocol.

The most restrictive setups of firewalls usually disable all incoming traffic, and only enable a few selected ports for outgoing traffic. In the solution below, it is assumed that one TCP port is enabled for outgoing connections to the GossipRouter.

JGroups has a mechanism that allows a programmer to tunnel a firewall. The solution involves a GossipRouter, which has to be outside of the firewall, so other members (possibly also behind firewalls) can access it.

The solution works as follows. A channel inside a firewall has to use protocol TUNNEL instead of UDP or TCP as transport. The recommended discovery protocol is PING. Here’s a configuration:

TUNNEL uses a GossipRouter (outside the firewall) running on HostA at port 12001 for tunneling. Note that it is not recommended to use TCPGOSSIP for discovery if TUNNEL is used (use PING instead). TUNNEL accepts one or multiple GossipRouters tor tunneling; they can be listed as a comma delimited list of host[port] elements specified in property gossip_router_hosts.

TUNNEL establishes a TCP connection to the GossipRouter process (outside the firewall) that accepts messages from members and passes them on to other members. This connection is initiated by the host inside the firewall and persists as long as the channel is connected to a group. A GossipRouter will use the same connection to send incoming messages to the channel that initiated the connection. This is perfectly legal, as TCP connections are fully duplex. Note that, if GossipRouter tried to establish its own TCP connection to the channel behind the firewall, it would fail. But it is okay to reuse the existing TCP connection, established by the channel.

Note that TUNNEL has to be given the hostname and port of the GossipRouter process. This example assumes a GossipRouter is running on HostA at port12001. TUNNEL accepts one or multiple router hosts as a comma delimited list of host[port] elements specified in property gossip_router_hosts.

Any time a message has to be sent, TUNNEL forwards the message to GossipRouter, which distributes it to its destination: if the message’s destination field is null (send to all group members), then GossipRouter looks up the members that belong to that group and forwards the message to all of them via the TCP connections they established when connecting to GossipRouter. If the destination is a valid member address, then that member’s TCP connection is looked up, and the message is forwarded to it.

A GossipRouter is not a single point of failure. In a setup with multiple gossip routers, the routers do not communicate among themselves, and a single point of failure is avoided by having each channel simply connect to multiple available routers. In case one or more routers go down, the cluster members are still able to exchange messages through any of the remaining available router instances, if there are any.

For each send invocation, a channel goes through a list of available connections to routers and attempts to send the message on each connection until it succeeds. If a message can not be sent on any of the connections, an exception is raised. The default policy for connection selection is random. However, we provide an plug-in interface for other policies as well.

The GossipRouter configuration is static and is not updated for the lifetime of the channel. A list of available routers has to be provided in the channel’s configuration file.

To tunnel a firewall using JGroups, the following steps have to be taken:

The general setup is shown in Tunneling a firewall:

First, the GossipRouter process is created on host B. Note that host B should be outside the firewall, and all channels in the same group should use the same GossipRouter process. When a channel on host A is created, its TCPGOSSIP protocol will register its address with the GossipRouter and retrieve the initial membership (assume this is C). Now, a TCP connection with the GossipRouter is established by A; this will persist until A crashes or voluntarily leaves the group. When A multicasts a message to the cluster, GossipRouter looks up all cluster members (in this case, A and C) and forwards the message to all members, using their TCP connections. In the example, A would receive its own copy of the multicast message it sent, and another copy would be sent to C.

This scheme allows for example Java applets , which are only allowed to connect back to the host from which they were downloaded, to use JGroups: the HTTP server would be located on host B and the gossip and GossipRouter daemon would also run on that host. An applet downloaded to either A or C would be allowed to make a TCP connection to B. Also, applications behind a firewall would be able to talk to each other, joining a group.

However, there are several drawbacks: first, having to maintain a TCP connection for the duration of the connection might use up resources in the host system (e.g. in the GossipRouter), leading to scalability problems, second, this scheme is inappropriate when only a few channels are located behind firewalls, and the vast majority can indeed use IP multicast to communicate, and finally, it is not always possible to enable outgoing traffic on 2 ports in a firewall, e.g. when a user does not own the firewall.

Here’s an example of the new configuration:

The attributes for the thread pools are prefixed with thread_pool respectively.

The attributes are listed below. They roughly correspond to the options of java.util.concurrent.ThreadPoolExecutor.

A message is considered delivered as soon as the receive() callback returns. While messages are received in a non-defined order, the reliable protocols (NAKACK2 and UNICAST3) establish an order in which messages are delivered.

Regular messages or message batches from a sender P are delivered in the order in which they were sent. E.g. if P sent messages 4 and 5, then the application’s receive() callback will be invoked with 4, and when 4 returns, with message 5. Alternatively, the application might receive a message batch containing messages 4 and 5. When iterating through that batch, message 4 will be consumed before message 5.

Regular messages from different senders P and Q are delivered in parallel. E.g if P sends 4 and 5 and Q sends 56 and 57, then the receive() callback might get invoked in parallel for P4 and Q57. Therefore the receive() callbacks have to be thread-safe.

In contrast, OOB messages are delivered in an undefined order, e.g. messages P4 and P5 might get delivered as P4 → P5 (P4 followed by P5) in some receivers and P5 → P4 in others. It is also possible that P4 is delivered in parallel with P5, each message getting delivered by a different thread.

The only guarantee for both regular and OOB messages is that a message will get delivered exactly once. Dropped messages are retransmitted and duplicate messages are dropped.

The following thread pools and factories are in TP:

The thread pool can be shared between instances running inside the same JVM. This can be done by creating an implementation of Executor, a number of channels and then setting the same executor in all channels via setThreadPool(Executor e).

The advantage here is that multiple channels running within the same JVM can pool (and therefore save) threads.

The disadvantage is that thread naming will not show to which channel instance an incoming thread belongs to.

JGroups creates all of its sockets through a SocketFactory, which is located in the transport (TP). The factory has methods to create sockets (Socket, ServerSocket, DatagramSocket and MulticastSocket), close sockets and list all open sockets. Every socket creation method has a service name, which could be for example "jgroups.fd_sock.srv_sock". The service name is used to look up a port (e.g. in a config file) and create the correct socket.

To provide one’s own socket factory, the following has to be done: the code below creates a SocketFactory implementation and sets it in the transport:

The application has to merge the substates from the various subgroups ({A,B,C} and {D,E}) back into one single state for {A,B,C,D,E}. This task has to be done by the application because JGroups knows nothing about the application state, other than it is a byte buffer.

If the in-memory state is backed by a database, then the solution is easy: simply discard the in-memory state and fetch it (eagerly or lazily) from the DB again. This of course assumes that the members of the 2 subgroups were able to write their changes to the DB. However, this is often not the case, as connectivity to the DB might have been severed by the network partition.

Another solution could involve tagging the state with time stamps. On merging, we could compare the time stamps for the substates and let the substate with the more recent time stamps win.

Yet another solution could increase a counter for a state each time the state has been modified. The state with the highest counter wins.

Again, the merging of state can only be done by the application. Whatever algorithm is picked to merge state, it has to be deterministic.

The primary partition approach is simple: on merging, one subgroup is designated as the primary partition and all others as non-primary partitions. The members in the primary partition don’t do anything, whereas the members in the non-primary partitions need to drop their state and re-initialize their state from fresh state obtained from a member of the primary partition.

The code to find the primary partition needs to be deterministic, so that all members pick the same primary partition. This could be for example the first view in the MergeView, or we could sort all members of the new MergeView and pick the subgroup which contained the new coordinator (the one from the consolidated MergeView). Another possible solution could be to pick the largest subgroup, and, if there is a tie, sort the tied views lexicographically (all Addresses have a compareTo() method) and pick the subgroup with the lowest ranked member.

Here’s code which picks as primary partition the first view in the MergeView, then re-acquires the state from the new coordinator of the combined view:

The handleView() method is called from viewAccepted(), which is called whenever there is a new view. It spawns a new thread which gets the subgroups from the MergeView, and picks the first subgroup to be the primary partition. Then, if it was a member of the primary partition, it does nothing, and if not, it reaqcuires the state from the coordinator of the primary partition (A).

The downside to the primary partition approach is that work (= state changes) on the non-primary partition is discarded on merging. However, that’s only problematic if the data was purely in-memory data, and not backed by persistent storage. If the latter’s the case, use state merging discussed above.

It would be simpler to shut down the non-primary partition as soon as the network partition is detected, but that a non trivial problem, as we don’t know whether {D,E} simply crashed, or whether they’re still alive, but were partitioned away by the crash of a switch. This is called a split brain syndrome, and means that none of the members has enough information to determine whether it is in the primary or non-primary partition, by simply exchanging messages.

In certain situations, we can avoid having multiple subgroups where every subgroup is able to make progress, and on merging having to discard state of the non-primary partitions.

If we have a fixed membership, e.g. the cluster always consists of 5 nodes, then we can run code on a view reception that determines the primary partition. This code

The algorithm is shown in pseudo code below:

Of course, the above mechanism requires that at least 3 nodes are up at any given time, so upgrades have to be done in a staggered way, taking only one node down at a time. In the worst case, however, this mechanism leaves the cluster read-only and notifies a system admin, who can fix the issue. This is still better than shutting the entire cluster down.

RELAY presents a global view to the application, e.g. a view received by nodes could be {D,E,F,A,B,C}. This view is the same on all nodes, and a global view is generated by taking the two local views, e.g. A|5 {A,B,C} and D|2 {D,E,F}, comparing the coordinators' addresses (the UUIDs for A and D) and concatenating the views into a list. So if D’s UUID is greater than A’s UUID, we first add D’s members into the global view ({D,E,F}), and then A’s members.

Therefore, we’ll always see all of A’s members, followed by all of D’s members, or the other way round.

To see which nodes are local and which ones remote, we can iterate through the addresses (PayloadUUID) and use the site (PayloadUUID.getPayload()) name to for example differentiate between "nyc" and "sfo".

To setup a relay, we need essentially 3 XML configuration files: 2 to configure the local clusters and 1 for the bridge.

To configure the first local cluster, we can copy udp.xml from the JGroups distribution and add RELAY on top of it: <RELAY bridge_props="/home/bela/tcp.xml" />. Let’s say we call this config relay.xml.

The second local cluster can be configured by copying relay.xml to relay2.xml. Then change the mcast_addr and/or mcast_port, so we actually have 2 different cluster in case we run instances of both clusters in the same network. Of course, if the nodes of one cluster are run in a different network from the nodes of the other cluster, and they cannot talk to each other, then we can simply use the same configuration.

The site property needs to be configured in relay.xml and relay2.xml, and it has to be different. For example, relay.xml could use site="nyc" and relay2.xml could use site="sfo".

The bridge is configured by taking the stock tcp.xml and making sure both local clusters can see each other through TCP.

If relay_multicasts is true then any multicast received by the site master of a site (ie. the coordinator of the local cluster, responsible for relaying of unicasts and multicasts) will relay the multicast to all connected sites. This means that - beyond setting relay_multicasts - nothing has to be done in order to relay multicasts across all sites.

A recipient of a multicast message which originated from a different site will see that the sender’s address is not a UUID, but a subclass (SiteUUID) which is the UUID plus the site suffix, e.g. A1:SFO. Since a SiteUUID is a subclass of a UUID, both types can be mixed and matched, placed into hashmaps or lists, and they implement compareTo() and equals() correctly.

When a reply is to be sent to the originator of the multicast message, Message.getSrc() provides the target address for the unicast response message. This is also a SiteUUID, but the sender of the response neither has to know nor take any special action to send the response, as JGroups takes care of routing the response back to the original sender.

As discussed above, relaying of unicasts is done transparently. However, if we don’t have a target address (e.g. as a result of reception of a multicast), there is a special address SiteMaster which identifies the site master; the coordinator of a local cluster responsible for relaying of messages.

Class SiteMaster is created with the name of a site, e.g. new SiteMaster("LON"). When a unicast with destination SiteMaster("LON") is sent, then we relay the message to the current site master of LON. If the site master changes, messages will get relayed to a different node, which took over the role of the site master from the old (perhaps crashed) site master.

Sometimes only certain members of a site should become site masters; e.g. the more powerful boxes (as routing needs some additional CPU power), or multi-homed hosts which are connected to the external network (over which the sites are connected with each other).

To do this, RELAY2 can generate special addresses which contain the knowledge about whether a member should be skipped when selecting a site master from a view, or not. If can_become_site_master is set to false in RELAY2, then the selection process will skip that member. However, if all members in a given view are marked with can_become_site_master=false, then the first member of the view will get picked.

When we have all members in a view marked with can_become_site_master=false, e.g. {B,C,D}, then B is the site master. If we now start a member A with can_become_site_master=true, then B will stop being the site master and A will become the new site master.

Invoking RPCs across sites is more or less transparent, except for the case when we cannot reach a member of a remote site. If we want to invoke method foo() in A1, A2 (local) and SiteMaster("SFO"), we could write the following code:

First, we add the members (A1 and A2) of the current (local) view to the destination set. Then we add the special address SiteMaster("SFO") which acts as a placeholder for the current coordinator of the SFO site.

Next, we invoke the call with dests as target set and block until responses from all A1, A2 and SiteMaster("SFO") have been received, or until 5 seconds have elapsed.

Next, we check the response list. And here comes the bit that’s new in 3.2: if a site is unreachable, a Rsp has an additional field "unreachable", which means that we could not reach the site master of SFO for example. Note that this is not necessarily an error, as a site maybe currently down, but the caller now has the option of checking on this new status field.

Let’s configure an example which consists of 3 sites SFO, LON and NYC and 2 members in each site. First we define the configuration for the local cluster (site) SFO. To do this, we could for example copy udp.xml from the JGroups distro (and name it sfo.xml) and add RELAY2 to the top (as shown above). RELAY2’s config property points to relay2.xml as shown above as well. The relay2.xml file defines a global cluster with global.xml, which uses TCP and MPING for the global cluster (copy for example tcp.xml to create global.xml)

Now copy sfo.xml to lon.xml and nyc.xml. The RELAY2 configuration stays the same for lon.xml and nyc.xml, but the multicast address and/or multicast port has to be changed in order to create 3 separate local clusters. Therefore, modify both lon.xml and nyc.xml and change mcast_port and / or mcast_addr in UDP to use separate values, so the clusters don’t interfere with each other.

To test whether we have 3 different clusters, start the Draw application (shipped with JGroups):

We should now have 3 local clusters (= sites) of 2 instances each. When RELAY2.relay_multicasts is true, if you draw in one instance, we should see the drawing in all 6 instances. This means that relaying of multicasting between sites works. If this doesn’t work, run a few Draw instances on global.xml, to see if they find each other.

Note that the first member of each cluster always joins the global cluster (defined by global.xml) too. This is necessary to relay messages between sites.

To test unicasts between sites, you can use the org.jgroups.demos.RelayDemoRpc program: start it as follows:

Start 2 instances in 3 sites and then use

to invoke RPCs on all local members and site masters SFO, NYC and LON. If one of the sites is down, you’ll get a message stating the site is unreachable.

It takes 6 rounds to broadcast m1 and m2 to the cluster.

In round 1, A send m1 to B.

In round 2, A sends m2 to B, but B also forwards m1 (received in round 1) to C.

In round 3, A is done. B forwards m2 to C and C forwards m1 to D (in parallel, denoted by ``||`).

In round 4, C forwards m2 to D.

Let’s take a look at this in terms of switch usage: in the N-1 approach, A can only send 125MB/sec, no matter how many members there are in the cluster, so it is constrained by the link capacity to the switch. (Note that A can also receive 125MB/sec in parallel with today’s full duplex links).

So the link between A and the switch gets hot.

In the daisychaining approach, link usage is more even: if we look for example at round 2, A sending to B and B sending to C uses 2 different links, so there are no constraints regarding capacity of a link. The same goes for B sending to C and C sending to D.

In terms of rounds, the daisy chaining approach uses X + (N-2) rounds, so for a cluster size of 10 and broadcasting 10 messages, it requires only 18 rounds, compared to 90 for the N-1 approach!

To measure performance of DAISYCHAIN, a performance test (test.Perf) was run, with 4 nodes connected to a 1 GB switch; and every node sending 1 million 8K messages, for a total of 32GB received by every node. The config used was tcp.xml.

The N-1 approach yielded a throughput of 73 MB/node/sec, and the daisy chaining approach 107MB/node/sec!

DAISYCHAIN can be placed directly on top of the transport, regardless of whether it is UDP or TCP, e.g.

MPerf is a test which measures multicast performance. This doesn’t mean IP multicast performance, but point-to-multipoint performance. Point-to-multipoint means that we measure performance of one-to-many messages; in other words, messages sent to all cluster members.

MPerf is dynamic; it doesn’t need a setup file to define the number of senders, number of messages to be sent and message size.

Instead, all the configuration needed by an instance of MPerf is an XML stack configuration, and configuration changes done in one member are automatically broadcast to all other members.

MPerf can be started as follows:

This assumes that we’re using IPv4 addresses (otherwise IPv6 addresses are used) and the JGroups JAR on the classpath.

A screen shot of MPerf looks like this (could be different, depending on the JGroups version):

We’re starting MPerf with -props ~/fast.xml and -name A. The -props option points to a JGroups configuration file, and -name gives the member the name "A".

A few instances of MPerf can now be started and each instance should join the same cluster.

MPerf can then be run by pressing [1]. In this case, every member in the cluster (in the example, we have members A and B) will send 1 million 1K messages. Once all messages have been received, MPerf will write a summary of the performance results to stdout:

In the sample run above, we see member A’s screen. A sends 1 million messages and waits for its 1 million and the 1 million messages from B to be received before it dumps some stats to stdout. The stats include the number of messages and bytes received, the time, the message rate and throughput averaged over the 2 members. It also shows the aggregated performance over the entire cluster.

In the sample run above (both processes on the same box), we got an average 756 MB of data per member per second, and an aggregated 1.5 GB per second for the entire cluster (A and B in this case).

Parameters such as the number of messages to be sent, the message size and the number of threads to be used to send the messages can be configured by pressing the corresponding numbers. After pressing return, the change will be broadcast to all cluster members, so that we don’t have to go to each member and apply the same change. Also, new members started, will fetch the current configuration and apply it.

For example, if we set the message size in A to 2000 bytes, then the change would be sent to B, which would apply it as well. If we started a third member C, it would also have a configuration with a message size of 2000.

Another feature is the ability to restart all cluster members with a new configuration. For example, if we modified ./fast.xml, we could select [6] to make all cluster members disconnect and close their existing channels and start a new channel based on the modified fast.xml configuration.

The new configuration file doesn’t even have to be accessible on all cluster members; only on the member which makes the change. The file contents will be read by that member, converted into a byte buffer and shipped to all cluster members, where the new channel will then be created with the byte buffer (converted into an input stream) as config.

Being able to dynamically change the test parameters and the JGroups configuration makes MPerf suited to be run in larger clusters; unless a new JGroups version is installed, MPerf will never have to be restarted manually.

UPerf is used to measure point-to-point (= unicast) communication between members. Start a few members like this:

They will form a cluster. When [1] is pressed, every node will invoke 20000 synchronous RPCs on other members, each time randomly selecting a member from the cluster. This will be done by 25 threads, but both number of RPCs and sender threads can be changed dynamically across the entire cluster at runtime.

With an 80% chance, a request will mimic a GET which is a small request returning a (by default) 1K response. With a 20% chance, the request is a PUT which is a 1K request and a small response. The read-write ration can be changed via [r].

GETs and PUTs mimic a distributed cache where GETs query information from the cache and PUT update information.

When done, every member sends its results back to the node on which the test was started, which then tallies the results, computes averages etc and prints the result of this round to stdout.

Here’s a sample run on member A:

This run was on a cluster consisting of {A,B,C,D} and the test was initiated on member A. When everyone is done, the results for A, B, C and D are printed individually, then averages for throughout and round-trip times are computed and also printed to stdout.

In this round, every node managed to invoke roughly 23'000 sync RPCs per second on randomly selected other members. The average GET time was slightly under 1 ms and PUT was roughly 1.6 ms.

Probe can also be used to inspect for every node P:

For sync RPCs, it is also possible to get the min/max/avg times for RPCs to a given destination.

Since taking the times for all sync RPCs takes time (2x System.nanoTime() for each RPC), this is disabled by default and has to be enabled (assuming we have 4 nodes running):

From now on, timings for sync RPCs will be taken (async RPCs are not timed and therefore not affected by the timing costs). To disable this, probe rpcs-disable-details can be called.

To get RPC stats, rpcs and rpcs-details can be used:

The example output shows stats for members C, A, B and D. When looking at the output of member A, we can see that A

To reset the numbers, probe.sh rpcs-reset can be called.

Probe uses IP multicasting by default. To enable TCP, diag_enable_tcp has to be set to true in the transport (and optionally diag_enable_udp to false).

Probe can then be started by pointing it to the address of one member, using the -addr argument, e.g.: probe.sh -addr 192.168.1.105 -port 7500 (note that port 7500 can be omitted, as it is the default). In this mode, probe will ask the member running at 192.168.1.105:7500 for a list of all other members, and then send the request to all of the returned members.

tshark is the command-line version of wireshark. To capture UDP packets for the configuration below, which has nodes binding to the loopback (127.0.0.1) device and ports starting from 7800, the following command can be used:

tshark -l -i en0 -i lo0 -T fields -e data udp and portrange 7800-7801 > jgroups-udp.data

This captures the data portion only (-T fields -e data) on the loopback and en0 devices and only captures packets sent to or being sent from ports 7800 and 7801.

To print the packets offline, java org.jgroups.tests.ParseMessages -file jgroups-udp.data can be used. This would look like this (edited):

This shows a typical conversation between A (coordinator) and B: B sends a discovery request, finds A and asks it to join (JOIN_REQ). A then multicasts (A to <all>) a view (VIEW) to all members, and sends a JOIN_RSP to B. Later, B leaves by sending a LEAVE_REQ to A and A responds with a LEAVE_RSP message back to B.

Instead of redirecting the output to a file, it could be piped into ParseMessages:

tshark -l -i en0 -i lo0 -T fields -e data udp and portrange 7800-7801 | java org.jgroups.tests.ParseMessages.

When we have captured a number of packets in wireshark, we can select "File" → "Export Specified Packets...":

Here, we can export selected number of packets, or all packets. In the example, the exported packets will be written to a file called jgroups-tcp.pcapng.

To parse this file and extract only the data sections, tshark can be used:

The -r options reads packets from the given input file and the -Tfields -edata options extract the data portion (containing the JGroups messages) only. This is piped to ParseMessages (with the -tcp option, as the wireshark data was captured from a TCP connection), which prints the JGroups messages to stdout.

tcpdump can be used with the -w <filename> command line option to save all captured packets to a file in PCAPNG (wireshark/libpcap) format.

In the example below (edited), however, we’re piping (options -U -w -) the captured packets to tshark which extracts the JGroups portion and in turn pipes this to ParseMessages:

Fork stacks can be created programmatically or declaratively. Let’s take a look at the latter first. The XML fragment below shows this:

FORK refers to an external file to configure its fork stacks:

The file fork-stacks.xml defines 2 fork stacks: "counter" and "lock". Each fork-stack element has an id attribute which defines the fork stack’s ID. Note that all fork stacks have to have unique IDs.

After the fork-stack element, the child element starting with config is a regular JGroups XML config file schema, where protocols are defined from bottom to top. For example, fork stack "lock" defines that CENTRAL_LOCK is the first protocol on top of FORK for the given fork stack, and STATS is on top of CENTRAL_LOCK.

When FORK is initialized, it will create the 2 fork stacks. When fork channels are created (see the next section), they can pick one of the 2 existing fork stacks to be created over, or they can dynamically create new fork stacks.

A fork channel is created by instantiating a new ForkChannel object:

First the main channel is created. Note that udp.xml may or may not contain FORK, but for this example, we assume it is present.

Then the ForkChannel is created. It is passed the main channel, the fork stack ID ("lock") and the fork channel ID ("fork-ch4"), plus a list of already instantiated protocols (CENTRAL_LOCK and STATS). If FORK already contains a fork stack with ID="lock", the existing fork stack will be used, or else a new one will be created with protocols CENTRAL_LOCK and STATS. Then a new fork channel with ID="fork-ch4" will be added to the top of fork stack "lock". An exception will be thrown if a fork channel with the same ID already exists.

The main channel then calls connect(), and - after this - the ForkChannel also calls connect(). (Note that the latter’s connect argument (the cluster name)) is ignored as fork channels have the same cluster name as the main channel they reference. The local address, name, view and state are also the same.

The lifetime of a fork channel is always dominated by the main channel: if the main channel is closed, all fork channels atttached to it are in closed state, too, and trying to send a message will throw an exception.

The example above showed the simplified constructor, which requires the FORK protocol to be present in the stack. There’s another constructor which allows for FORK to be created dynamically if not present:

In addition to passing the main channel, the fork stack and channel IDs and the list of protocols, this constructor also allows a user to create FORK in the main stack if not present. To do so, create_fork_if_absent has to be set to true (else an exception is thrown if FORK is not found), and the neighbor protocol (e.g. FRAG2.class) has to be defined, plus the position (ProtocolStack.ABOVE/BELOW) relative to the neighbor protocol has to be defined as well.

The design of FORK / ForkChannel is discussed in more detail in FORK.txt

UDP uses IP multicast for sending messages to all members of a group and UDP datagrams for unicast messages (sent to a single member). When started, it opens a unicast and multicast socket: the unicast socket is used to send/receive unicast messages, whereas the multicast socket sends and receives multicast messages. The channel’s physical address will be the address and port number of the unicast socket. A protocol stack with UDP as transport protocol is typically used with clusters whose members run in the same subnet. If running across subnets, an admin has to ensure that IP multicast is enabled across subnets. It is often the case that IP multicast is not enabled across subnets. In such cases, the stack has to either use UDP without IP multicasting or other transports such as TCP.

Specifying TCP in your protocol stack tells JGroups to use TCP to send messages between cluster members. Instead of using a multicast bus, the cluster members create a mesh of TCP connections. For example, while UDP sends 1 IP multicast packet when sending a message to a cluster of 10 members, TCP needs to send the message 9 times. It sends the same message to the first member, to the second member, and so on (excluding itself as the message is looped back internally).

This is slow, as the cost of sending a group message is O(n) with TCP, where it is O(1) with UDP. As the cost of sending a group message with TCP is a function of the cluster size, it becomes higher with larger clusters.

TCP_NIO2 is similar to TCP, but uses NIO (= Non blocking IO) to send messages to and receive messages from members. Contrary to TCP, it doesn’t use 1 thread per connection, but handles accepts, connects, reads and writes in a single thread.

All of these operations are guaranteed to never block.

For example, if a read is supposed to receive 1000 bytes and only reveived 700, the read reads the 700 bytes, saves them somewhere and later - when the remaining 300 bytes have been received - is notified to complete the read and then returns the 1000 bytes to the application.

Using a single thread is not a problem, as operations will never block. The only potentially blocking operation, namely delivering messages up to the application, is done via the regular or OOB thread pools, as usual.

While TCP and TCP_NIO2 both have the N-1 problem of sending cluster wide messages (contrary to UDP), TCP_NIO2 is able to handle a larger number of connections than TCP, as it doesn’t use the thread-per-connection model, and - contrary to TCP, but similar to UDP - it doesn’t block when sending or receiving messages.

TUNNEL is described in TUNNEL.

Discovery is the superclass for all discovery protocols and therefore its properties below can be used in any subclass.

Discovery sends a discovery request, and waits for num_initial_members discovery responses, or timeout ms, whichever occurs first, before returning. Note that break_on_coord_rsp="true" will return as soon as we have a response from a coordinator.

Initial (dirty) discovery of members. Used to detect the coordinator (oldest member), by mcasting PING requests to an IP multicast address. Each member responds with a packet {C, A}, where C=coordinator’s address and A=own address. After N milliseconds or M replies, the joiner determines the coordinator from the responses, and sends a JOIN request to it (handled by GMS). If nobody responds, we assume we are the first member of a group. Unlike TCPPING, PING employs dynamic discovery, meaning that the member does not have to know in advance where other cluster members are. PING uses the IP multicasting capabilities of the transport to send a discovery request to the cluster. It therefore requires UDP as transport.

TCPPING is used with TCP as transport, and uses a static list of cluster members’s addresses. See Using TCP and TCPPING for details.

TCPGOSSIP uses an external GossipRouter to discover the members of a cluster. See Using TCP and TCPGOSSIP for details.

MPING (=Multicast PING) uses IP multicast to discover the initial membership. It can be used with all transports, but usually is used in combination with TCP. TCP usually requires TCPPING, which has to list all cluster members explicitly, but MPING doesn’t have this requirement. The typical use case for this is when we want TCP as transport, but multicasting for discovery so we don’t have to define a static list of initial hosts in TCPPING MPING uses its own multicast socket for discovery. Properties bind_addr (can also be set via -Djgroups.bind_addr=), mcast_addr and mcast_port can be used to configure it. Note that MPING requires a separate thread listening on the multicast socket for discovery requests.

FILE_PING can be used instead of GossipRouter in cases where no external process is desired.

Since 3.5, the way FILE_PING performs discovery has changed. The following paragraphs describe the new mechanism to discover members via FILE_PING or subclasses (e.g. S3_PING GOOGLE_PING or GOOGLE_PING2), so this applies to all cloud-based stores as well. Instead of storing 1 file per member in the file system or cloud store, we only store 1 file for all members. This has the advantage, especially in cloud stores, that the number of reads is not a function of the cluster size, e.g. we don’t have to perform 1000 reads for member discovery in a 1000 node cluster, but just a single read.

This is important as the cost of 1000 times the round trip time of a (REST) call to the cloud store is certainly higher that the cost of a single call. There may also be a charge for calls to the cloud, so a reduced number of calls lead to reduced charges for cloud store access, especially in large clusters.

The current coordinator is always in charge of writing the file; participants never write it, but only read it. When there is a split and we have multiple coordinator, we may also have multiple files.

The name of a file is always UUID.logical_name.list, e.g. 0000-0000-000000000001.m1.1.list, which has a UUID of 1, a logical name of "m1.1" and the suffix ".list".

JDBC_PING uses a DB to store information about cluster nodes used for discovery. All cluster nodes are supposed to be able to access the same DB.

When a node starts, it queries information about existing members from the database, determines the coordinator and then asks the coord to join the cluster. It also inserts information about itself into the table, so others can subsequently find it.

When a node P has crashed, the current coordinator removes P’s information from the DB. However, if there is a network split, then this can be problematic, as crashed members cannot be told from partitioned-away members.

For instance, if we have {A,B,C,D}, and the split creates 2 subclusters {A,B} and {C,D}, then A would remove {C,D} because it thinks they crashed, and - likewise - C would remove {A,B}.

To solve this, every member re-inserts its information into the DB after a view change. So when C and D's view changes from {A,B,C,D} to {C,D}, both sides of the split re-insert their information. Ditto for the other side of the network split.

The re-insertion is governed by attributes info_writer_max_writes_after_view and info_writer_sleep_time: the former defines the number of times re-insertion should be done (in a timer task) after each view change and the latter is the sleep time (in ms) between re-insertions.

The value of this is that dead members are removed from the DB (because they cannot do re-insertion), but network splits are handled, too.

Another attribute clear_table_on_view_change governs how zombies are handled. Zombies are table entries for members which crashed, but weren’t removed for some reason. E.g. if we have a single member A and kill it (via kill -9), then it won’t get removed from the table.

If clear_table_on_view_change is set to true, then the coordinator clears the table after a view change (instead of only removing the crashed members), and everybody re-inserts its own information. This attribute can be set to true if automatic removal of zombies is desired. However, it is costly, therefore if no zombies ever occur (e.g. because processes are never killed with kill -9), or zombies are removed by a system admin, then it should be set to false.

BPING uses UDP broadcasts to discover other nodes. The default broadcast address (dest) is 255.255.255.255, and should be replaced with a subnet specific broadcast, e.g. 192.168.1.255.

RACKSPACE_PING uses Rackspace Cloud Files Storage to discover initial members. Each node writes a small object in a shared Rackspace container. New joiners read all addresses from the container and ping each of the elements of the resulting set of members. When a member leaves, it deletes its corresponding object.

This objects are stored under a container called jgroups, and each node will write an object name after the cluster name, plus a "/" followed by the address, thus simulating a hierarchical structure.

S3_PING uses Amazon S3 to discover initial members. New joiners read all addresses from this bucket and ping each of the elements of the resulting set of members. When a member leaves, it deletes its corresponding file.

It’s designed specifically for members running on Amazon EC2, where multicast traffic is not allowed and thus MPING or PING will not work. When Amazon RDS is preferred over S3, or if a shared database is used, an alternative is to use JDBC_PING.

Each instance uploads a small file to an S3 bucket and each instance reads the files out of this bucket to determine the other members.

There are three different ways to use S3_PING, each having its own tradeoffs between security and ease-of-use. These are described in more detail below:

Here’s a configuration example for private buckets with credentials given to each instance:

Here’s an example for public buckets with no credentials:

And finally, here’s an example for public readable buckets with pre-signed URLs:

This is a protocol written by Meltmedia, which uses the AWS API. It is not part of JGroups, but can be downloaded at https://github.com/meltmedia/jgroups-aws.

This implementation by Zalando uses the AWS SDK. It is not part of JGroups, but can be found at https://github.com/zalando/jgroups-native-s3-ping. This protocols works with JGroups versions 3.x.

There’s a refactored version of AWS_PING that was ported (in 2017) to run on JGroups 4.x at https://github.com/jgroups-extras/native-s3-ping.

GOOGLE_PING is a subclass of S3_PING and inherits most of the functionality. It uses Google Cloud Storage to store information about individual members.

The snippet below shows a sample config:

This will use a bucket "jgroups-bucket" or create one if it doesn’t exist, then create another folder under it with the cluster name, and finally use 1 object per member in that location for member info.

GOOGLE_PING2 is the successor to GOOGLE_PING and uses Google’s client library to access Google Compute Storage. It is the recommended way to access GCS and the project is hosted at https://github.com/jgroups-extras/jgroups-google.

DNS_PING uses DNS A or SRV entries to perform discovery. Initially this protocol was designed for Kubernetes and OpenShift but it suitable for any type of DNS discovery.

The snippet below shows a sample config:

This will turn on DNS discovery using the DNS server at address 192.168.0.17 and DNS query jgroups-dns-ping.myproject.svc.cluster.local using DNS A records.

The dns_address parameter is optional and when it’s missing, the protocol will use the default DNS resolver configured on the machine.

The dns_query parameter is mandatory. It is used for querying the DNS Server and obtaining information about the cluster members. The svc.cluster.local part is specific to Kubernetes and OpenShift and might be omitted.

It is also possible to use SRV entries for discovery as shown below:

Kubernetes SRV entries are created using the following scheme: _my-port-name._my-port-protocol.my-svc.my-namespace.svc.cluster.local.

When the above example is used in Kubernetes or OpenShift, [DNS_PING] will form a cluster of all the pods governed by a service named jgroups-dns-ping in namespace myproject, which exposes a TCP port named ping.

Here’s an example of a YAML file which shows how to run a service and a pod using DNS_PING:

Configuration dns_ping.xml sets up DNS_PING as follows:

The DNS_QUERY system property (overriding the dns_query attribute) is defined in the Yaml as _ping._tcp.jgrp.default.svc.cluster.local., which corresponds to the port (7800) advertized in the ports section: _ping is the port name, _tcp the protocol, jgrp the project and default the namespace.

As can also be seen in the Yaml file, DNS_RECORD_TYPE is set to SRV, overriding the default type of A.

For more information, please refer to Kubernetes DNS Admin Guide.

A working example of using this protocol might be found in https://github.com/slaskawi/jgroups-dns-ping-example.

SWIFT_PING uses Openstack Swift to discover initial members. Each node writes a small object in a shared container. New joiners read all addresses from the container and ping each of the elements of the resulting set of members. When a member leaves, it deletes its corresponding object.

These objects are stored under a container called jgroups (by default), and each node will write an object name after the cluster name, plus a "/" followed by the address, thus simulating a hierarchical structure.

Currently only Openstack Keystone authentication is supported. Here is a sample configuration block:

This Kubernetes-based discovery protocol can be used with OpenShift [2] and uses Kubernetes to discover cluster members. KUBE_PING is hosted on jgroups-extras; refer to [1] for details.

[1] https://github.com/jgroups-extras/jgroups-kubernetes

[2] https://www.openshift.com

This is a discovery protocol that allows cluster nodes to run on the Azure cloud [1]. For details refer to [2].

[1] https://azure.microsoft.com/en-us/

[2] https://github.com/jgroups-extras/jgroups-azure

The Persistent Discovery Cache can be used to cache the results of the discovery process persistently. E.g. if we have TCPPING.initial_hosts configured to include only members A and B, but have a lot more members, then other members can bootstrap themselves and find the right coordinator even when neither A nor B are running.

An example of a TCP-based stack configuration is:

This discovery protocol allows for multiple discovery protocols to be used in a configuration. MULTI_PING adds all discovery protocols underneath it in the stack to a list at initialization time.

When a discovery request is received from above, all discovery protocols in the list are contacted, either sequentially (async_discovery=false) or in parallel (async_discovery=true).

A sample configuration is shown below:

If a cluster gets split for some reasons (e.g. network partition), this protocol merges the subclusters back into one cluster.

All members periodically send an INFO message with their address (UUID), logical name, physical address and ViewId. The ViewId (ViewId) is used to see if we have diverging views among the cluster members: periodically, every coordinator looks at the INFO messages received so far and checks if there are any inconsistencies.

If inconsistencies are found, the merge leader will be the member with the lowest address (UUID).

The merge leader then asks the senders of the inconsistent ViewIds for their full views. Once received, it simply passes a MERGE event up the stack, where the merge will be handled (by GMS) in exactly the same way as if MERGE2 has generated the MERGE event.

The advantages of MERGE3 are:

Failure detection based on a logical ring and heartbeat messages.

Members form a logical ring; e.g. in view {A,B,C,D}, A pings B, which pings C, which pings D, which pings A. Pinging means sending a heartbeat.

Each member sends this heartbeat every timeout ms to the neighbor to its right. When a member receives a heartbeat, it sends back an ack. When the ack is received the timestamp of when a member last heard from its neighbor is reset.

When a member doesn’t receive any heartbeat acks from its neighbor for timeout * max_tries ms, that member is declared suspected, and will be excluded by GMS.

This is done by FD multicasting a SUSPECT(P) message which is handled by the current coordinator by double-checking the health of P (using VERIFY_SUSPECT) and - if P still doesn’t reply - by excluding P from the membership.

Note that setting msg_counts_as_heartbeat in P to true causes the timestamp of P in the pinging member to be reset.

Failure detection based on simple heartbeat protocol. Every member periodically multicasts a heartbeat. Every member also maintains a table of all members (minus itself). When data or a heartbeat from P are received, we reset the timestamp for P to the current time. Periodically, we check for expired members whose timestamp is greater than the timeout, and suspect those.

Similar to FD_ALL, but doesn’t use any timestamps. Instead, a boolean flag is associated with each member. When a message or heartbeat (sent every interval ms) from P is received, P’s flag is set to true. The heartbeat checker checks every timeout ms for members whose flag is false, suspects those, and - when done - resets all flags to false again. The times it takes to suspect a member are the same as for FD_ALL

Failure detection protocol which maintains a bitmap of timeout / interval bits (e.g. timeout=60000, interval=10000 → 6 bits), initialized to 1, for each member. The timeout check task also maintains an index which is incremented every time it is invoked and the bit at the index is set to 0.

When a heartbeat or a message is received, the bit at the current index is set to 1.

When the timeout check task detects that all bits are 0, the member will be suspected.

Failure detection protocol based on a ring of TCP sockets created between cluster members, similar to FD but not using heartbeat messages.

Each member in a cluster connects to its neighbor (the last member connects to the first), thus forming a ring. Member B is suspected when its neighbor A detects abnormal closing of its TCP socket (presumably due to a crash of B). However, if B is about to leave gracefully, it lets its neighbor A know, so that A doesn’t suspect B. ===== Example * The membership is {A,B,C,D,E}. * Members C and D are killed at the same time * B notices that C abnormally closed its TCP socket and broadcasts a SUSPECT(C) message * The current coordinator (A) asks VERIFY_SUSPECT to double check that C is dead * Meanwhile, B tries to create a TCP socket to the next-in-line (D) but fails. It therefore broadcasts a SUSPECT(D) message * A also handles this message and asks VERIFY_SUSPECT to double check if D is dead * After VERIFY_SUSPECT can’t verify that C and D are still alive, A creates a new view {A,B,E} and installs it * The time taken for FD_SOCK to suspect a member is very small (a few ms)

To detect the crash or freeze of entire hosts and all of the cluster members running on them, FD_HOST can be used. It is not meant to be used in isolation, as it doesn’t detect crashed members on the local host, but in conjunction with other failure detection protocols, such as FD_ALL or FD_SOCK.

FD_HOST can be used when we have multiple cluster members running on a physical box. For example, if we have members {A,B,C,D} running on host 1 and {M,N,O,P} running on host 2, and host 1 is powered down, then A, B, C and D are suspected and removed from the cluster together, typically in one view change.

By default, FD_HOST uses InetAddress.isReachable() to perform liveness checking of other hosts, but if property cmd is set, then any script or command can be used. FD_HOST will launch the command and pass the IP address ot the host to be checked as argument. Example: cmd="ping -c 3".

A typical failure detection configuration would look like this:

If we have members {A,B,C} on host 192.168.1.3, {M,N,O} on 192.168.1.4 and {X,Y,Z} on 192.168.1.5, then the behavior is as follows:

Verifies that a suspected member is really dead by pinging that member one last time before excluding it, and dropping the suspect message if the member does respond.

VERIFY_SUSPECT tries to minimize false suspicions.

The protocol works as follows: it catches SUSPECT events traveling up the stack. Then it verifies that the suspected member is really dead. If yes, it passes the SUSPECT event up the stack, otherwise it discards it. VERIFY_SUSPECT Has to be placed somewhere above the failure detection protocol and below the GMS protocol (receiver of the SUSPECT event). Note that SUSPECT events may be reordered by this protocol.

NAKACK2 provides reliable delivery and FIFO (= First In First Out) properties for messages sent to all nodes in a cluster.

It performs lossless and FIFO delivery of multicast messages, using negative acks. E.g. when receiving P:1, P:3, P:4, a receiver delivers only P:1, and asks P for retransmission of message 2, queuing P3-4. When P2 is finally received, the receiver will deliver P2-4 to the application.

Reliable delivery means that no message sent by a sender will ever be lost, as all messages are numbered with sequence numbers (by sender) and retransmission requests are sent to the sender of a message if that sequence number is not received.

FIFO order means that all messages from a given sender are received in exactly the order in which they were sent.

UNICAST3 provides reliable delivery and FIFO (= First In First Out) properties for point-to-point messages between a sender and a receiver.

Reliable delivery means that no message sent by a sender will ever be lost, as all messages are numbered with sequence numbers (by sender) and retransmission requests are sent to the sender of a message if that sequence number is not received. UNICAST3 uses a mixture of positive and negative acks (similar to NAKACK2). This reduces the communication overhead required for sending an ack for every message.

FIFO order means that all messages from a given sender are received in exactly the order in which they were sent.

On top of a reliable transport, such as TCP, UNICAST3 is not really needed. However, concurrent delivery of messages from the same sender is prevented by UNICAST3 by acquiring a lock on the sender’s retransmission table, so unless concurrent delivery is desired, UNICAST3 should not be removed from the stack even if TCP is used.

Details of UNICAST3’s design can be found here: UNICAST3

The RSVP protocol is not a reliable delivery protocol per se, but augments reliable protocols such as NAKACK, UNICAST or UNICAST2. It should be placed somewhere above these in the stack.

STABLE garbage collects messages that have been seen by all members of a cluster. Each member has to store all messages because it may be asked to retransmit. Only when we are sure that all members have seen a message can it be removed from the retransmission buffers. STABLE periodically gossips its highest and lowest messages seen. The lowest value is used to compute the min (all lowest seqnos for all members), and messages with a seqno below that min can safely be discarded.

Note that STABLE can also be configured to run when N bytes have been received. This is recommended when sending messages at a high rate, because sending stable messages based on time might accumulate messages faster than STABLE can garbage collect them.

For example, if no unicast flow control is needed, UFC can be left out of the stack configuration.

This is the non-blocking alternative to UFC. It extends UFC, so all attributes from UFC are inherited.

This is the non-blocking alternative to MFC. It inherits from MFC, so all attributes are inherited.

FRAG and FRAG2 fragment large messages into smaller ones, send the smaller ones, and at the receiver side, the smaller fragments will get assembled into larger messages again, and delivered to the application. FRAG and FRAG2 work for both unicast and multicast messages.

The difference between FRAG and FRAG2 is that FRAG2 does 1 less copy than FRAG, so it is the recommended fragmentation protocol. FRAG serializes a message to know the exact size required (including headers), whereas FRAG2 only fragments the payload (excluding the headers), so it is faster.

The properties of FRAG2 are:

Contrary to FRAG, FRAG2 does not need to serialize a message in order to break it into smaller fragments: it looks only at the message’s buffer, which is a byte array anyway. We assume that the size addition for headers and src and dest addresses is minimal when the transport finally has to serialize the message, so we add a constant (by default 200 bytes). Because of the efficiency gained by not having to serialize the message just to determine its size, FRAG2 is generally recommended over FRAG. .FRAG3

FRAG2 needs only half the memory than FRAG2 to handle fragments and the final full message. See https://issues.jboss.org/browse/JGRP-2154 for details.

SEQUENCER provider total order for multicast (=group) messages by forwarding messages to the current coordinator, which then sends the messages to the cluster on behalf of the original sender. Because it is always the same sender (whose messages are delivered in FIFO order), a global (or total) order is established.

Sending members add every forwarded message M to a buffer and remove M when they receive it. Should the current coordinator crash, all buffered messages are forwarded to the new coordinator.

A total order anycast is a totally ordered message sent to a subset of the cluster members. TOA intercepts messages with an AnycastMessage (carrying a list of addresses) and handles sending of the message in total order. Say the cluster is {A,B,C,D,E} and the Anycast is to {B,C}.

Skeen’s algorithm is used to send the message: B and C each maintain a logical clock (a counter). When a message is to be sent, TOA contacts B and C and asks them for their counters. B and C return their counters (incrementing them for the next request).

The originator of the message then sets the message’s ID to be the max of all returned counters and sends the message. Receivers then deliver the messages in order of their IDs.

The main use of TOA is currently in Infinispan’s transactional caches with partial replication: it is used to apply transactional modifications in total order, so that no two-phase commit protocol has to be run and no locks have to be acquired.

As shown in "Exploiting Total Order Multicast in Weakly Consistent Transactional Caches", when we have many conflicts by different transactions modifying the same keys, TOM fares better than 2PC.

Note that TOA is experimental (as of 3.1).

STATE_TRANSFER is the existing transfer protocol, which transfers byte[] buffers around. However, at the state provider’s side, JGroups creates an output stream over the byte[] buffer, and passes the ouput stream to the getState(OutputStream) callback, and at the state requester’s side, an input stream is created and passed to the setState(InputStream) callback.

This allows us to continue using STATE_TRANSFER, until the new state transfer protocols are going to replace it (perhaps in 4.0).

In order to transfer application state to a joining member of a cluster, STATE_TRANSFER has to load entire state into memory and send it to a joining member. The major limitation of this approach is that for state transfers that are very large this would likely result in memory exhaustion.

For large state transfer use either the STATE or STATE_SOCK protocol. However, if the state is small, STATE_TRANSFER is okay.

StreamingStateTransfer is the superclass of STATE and STATE_SOCK (see below). Its properties are:

STATE_SOCK is also a streaming state transfer protocol, but compared to STATE, it doesn’t send the chunks as messages, but uses a TCP socket connection between state provider and requester to transfer the state.

The state provider creates a server socket at a configurable bind address and port, and the address and port are sent back to a state requester in the state response. The state requester then establishes a socket connection to the server socket and passes the socket’s input stream to the setState(InputStream) callback.

BARRIER is used by some of the state transfer protocols, as it lets existing threads complete and blocks new threads to get both the digest and state in one go.

In 3.1, a new mechanism for state transfer will be implemented, eliminating the need for BARRIER. Until then, BARRIER should be used when one of the state transfer protocols is used. BARRIER is part of every default stack which contains a state transfer protocol.

Encryption is based on a shared secret key that all members of a cluster have. The key is either acquired from a shared keystore (symmetric encryption) or a new joiner fetches it from the coordinator via public/private key exchange (asymmetric encryption).

A sender encrypts a message with the shared secret key and the receivers decrypt it with the same secret key.

By default, only the payload of a message is encrypted, but not the other metadata (e.g. headers, destination address, flags etc).

If (for example) headers are not encrypted, it is possible to use replay attacks, because the sequence number (seqno) of a message is seen. For example, if a seqno is 50, then an attacker might copy the message, and increment the seqno.

To prevent this, the SERIALIZE protocol can be placed on top of SYM_ENCRYPT or ASYM_ENCRYPT. It serializes the entire message into the payload of a new message that’s then encrypted and sent down the stack.

Authentication is performed by AUTH. Its main use is to make sure only authenticated members can join (or merge into) a cluster. Scenarios where AUTH kicks in are:

So authentication makes sure that rogue nodes will never be able to be members of a cluster, be it via joining or merging.

AUTH provides pluggable security that defines if a node should be allowed to join a cluster. It can be used standalone, or in conjunction with encryption protocols such as SYM_ENCRYPT or ASYM_ENCRYPT.

AUTH sits below the GMS protocol and listens for JOIN REQUEST messages. When a JOIN REQUEST is received it tries to find an AuthHeader object, inside of which should be an implementation of the AuthToken object.

AuthToken is an abstract class, implementations of which are responsible for providing the actual authentication mechanism. Some basic implementations of AuthToken are provided in the org.jgroups.auth package (e.g X509Token, FixedMembershipToken etc).

If authentication is successful, the message is simply passed up the stack to the GMS protocol.

If it fails, the AUTH protocol creates a JOIN RESPONSE message with a failure string and passes it back down the stack. This failure string informs the client of the reason for failure. Clients will then fail to join the group and will throw a SecurityException. If this error string is null then authentication is considered to have passed.

For historical (= outdated) information refer to the wiki at AUTH.

SASL is an alternative to the AUTH protocol which provides a layer of authentication to JGroups by allowing the use of one of the SASL mechanisms made available by the JDK. SASL sits below the GMS protocol and listens for JOIN / MERGE REQUEST messages. When a JOIN / MERGE REQUEST is received it tries to find a SaslHeader object which contains the initial response required by the chosen SASL mech. This initiates a sequence of challenge/response messages which, if successful, culminates in allowing the new node to join the cluster. The actual validation logic required by the SASL mech must be provided by the user in the form of a standard javax.security.auth.CallbackHandler implementation.

When authentication is successful, the message is simply passed up the stack to the GMS protocol. When it fails, the SASL protocol creates a JOIN / MERGE RESPONSE message with a failure string and passes it back down the stack. This failure string informs the client of the reason for failure. Clients will then fail to join the group and will throw a SecurityException. If this error string is null then authentication is considered to have passed.

SASL can be (minimally) configured as follows:

The mech property specifies the SASL mech you want to use, as defined by RFC-4422. You will also need to provide two callback handlers, one used when the node is running as coordinator (server_callback_handler) and one used in all other cases (client_callback_handler). Refer to the JDK’s SASL reference guide for more details: http://docs.oracle.com/javase/7/docs/technotes/guides/security/sasl/sasl-refguide.html

The JGroups package comes with a simple properties-based CallbackHandler which can be used when a more complex Kerberos/LDAP approach is not needed. To use this set both the (server_callback_handler) and the (client_callback_handler) to org.jgroups.auth.sasl.SimpleAuthorizingCallbackHandler. This CallbackHandler can be configured either programmatically by passing to the constructor an instance of java.util.Properties containing the appropriate properties, or via standard Java system properties (i.e. set on the command-line using the -DpropertyName=propertyValue notation. The following properties are available: