Fine-Tuning WAN Replication

The maximum size of events that are sent in a single batch can be changed depending on your needs. The batch of events is not sent until this size is reached or enough time has elapsed. The default value for batch size is 500. The batch size can be set for each WAN publisher separately by modifying the related WanBatchPublisherConfig.

Below is the configuration for changing the value of the element:

When using the built-in WAN batch replication, if the number of WAN replication events generated does not reach Batch Size, they are sent to the target cluster after a certain amount of time is passed. You can set this duration in milliseconds using this batch maximum delay configuration. Default value of for this duration is 1 second (1000 milliseconds).

Maximum delay can be set for each target cluster by modifying related WanBatchPublisherConfig.

You can change this element using the configuration as shown below.

After a replication event is sent to the target cluster, the source member waits for an acknowledgement of the delivery of the event to the target. If the confirmation is not received inside a timeout duration window, the event is resent to the target cluster. Default value of this duration is 60000 milliseconds.

You can change this duration depending on your network latency for each target cluster by modifying related WanBatchPublisherConfig.

Below is an example configuration:

For clusters with high data mutation rates or with long expected periods of disrupted connectivity between clusters, you might need to increase the replication queue size. The default queue size for replication queues is 10000. This means, if you have heavy put/update/remove rates or if the target/passive cluster is unavailable for too long, you might exceed the queue size so that the oldest, not yet replicated, updates might get lost. Note that a separate queue is used for each WAN Replication configured for IMap and ICache.

Queue capacity can be set for each target cluster by modifying the related WanBatchPublisherConfig.

You can change this element using the configuration as shown below.

Note that you can clear a member’s WAN replication event queue. It can be initiated through Management Center’s Clear Queues action or Hazelcast’s REST API. Below is the URL for its REST call:

You need to add the following URL-encoded parameters to the request in the following order separated by "&";

This may be useful, for instance, to release the consumed heap if you know that the target cluster is being shut down, decommissioned, put out of use and it will never come back. Or, when a failure happens and queues are not replicated anymore, you could clear the queues using this clearing action.

You can also configure the policy to be applied when the WAN Replication event queues are full. The following policies are supported:

The following is an example configuration:

WAN replication supports different acknowledgment (ACK) types for each target cluster. You can choose from two different acknowledgment types depending on your consistency and performance requirements. The following ACK types are supported:

The following is an example configuration:

By default, WAN Replication will replicate all the updates on map and cache entries. If you are updating a single "hot" entry multiple times, WAN Replication will send an update event for every entry update. If you don’t need to have all updates replicated and would like to simply replicate the latest update for a certain entry, you can turn on key-based coalescing, thus saving on amounts of data replicated between clusters.

The following is an example configuration:

The WAN replication mechanism allows tuning for lower latencies of replication and higher throughput. In most cases, WAN replication is sufficient with out-of-the-box settings which cause WAN replication to replicate the map and cache events with little overhead. However, there might be some use cases where the latency between a map/cache mutation on one cluster and its visibility on the other cluster must be kept within some bounds. To achieve such demands, you can first try tuning the WAN replication mechanism using the following publisher elements:

To understand the implications of these elements, let’s first dive into how WAN replication works.

WAN replication runs in a separate thread and tries to send map and cache mutation events in batches to the target endpoints for higher throughput. The target endpoints are usually members in a target Hazelcast cluster but different WAN implementations may have different target endpoints. The event batch is collected by iterating over the WAN queues for different partitions and, different maps and caches. WAN replication tries and collects a batch of a size which can be configured using the batch-size element.

If enough time has passed and the WAN replication thread hasn’t collected enough events to fill a batch, it sends what it has collected nevertheless. This is controlled by the batch-max-delay-millis element. The "enough time" precisely means that more than the configured amount of milliseconds has passed since the time the last batch was sent to any target endpoint.

If there are no events in any of the WAN queues, the WAN replication thread goes into the idle state by parking the WAN replication thread. The minimum park time can be defined using the idle-min-park-ns element and the maximum park time can be controlled using the idle-max-park-ns element. If a WAN event is enqueued while the WAN replication thread is in the idle state, the latency for replication of that WAN event increases.

An example WAN replication configuration using the default values of the above elements is shown below.

We will now discuss tuning these elements. Unfortunately, the exact tuning parameters heavily depend on the load, mutation rate, latency between the source and target clusters and even use cases. We will thus discuss some general approaches and pointers.

When tuning for low latency, the first thing you might want to do is lower the idle-min-park-ns and idle-max-park-ns element values. This will affect the latencies that you see when having a low number of operations per second, since this is when the WAN replication thread will be mostly in idle state. Try lowering both elements but keep in mind that the lower the element value, the more time the WAN replication thread will spend consuming CPU in a quiescent state - when there is no mutation on the maps or caches.

The next element you might lower is the batch-max-delay-millis. If you have a strict upper bound on the latency for WAN replication, this element must be below that limit. Setting this value too low might adversely affect the performance: in that case the WAN replication thread might send smaller batches than what it would if the element was higher and it had waited for some more time. You can even try setting this element to zero which instructs the WAN replication thread to send batches as soon as it is able to collect any events; but keep in mind this will result in many smaller batches instead of fewer larger event batches.

When tuning for lower latencies, configuring the batch-size usually has little effect, especially at lower mutation rates. At a low number of operations per second, the event batches will usually be very small since the WAN replication thread will not be able to collect the full batch and respect the required latencies for replication. The batch-size element might have more effect at higher mutation rates. Here, you will probably want to use bigger batches to avoid paying for the latencies when sending lots of smaller batches, so try increasing the batch size and benchmarking at high load.

There are a couple of other configuration values that you might try changing but it depends on your use case. The first one is adding a separate configuration for a WAN replication executor. Collecting of WAN event batches and processing the responses from the target endpoints are done in a thread pool shared between the other parts of the Hazelcast system and all WAN replication schemes. In some cases, you might want to define how many threads WAN replication can use by setting the thread pool size of WAN’s dedicated executor. The name of this executor is hz:wan. Below is an example of a concrete, dedicated executor for WAN replication. See the Configuring Executor Service section for more information about the configuration options of the executor.

In this case, WAN replication is limited to use 16 threads at most, which is the default. Note that every configured WAN batch publisher occupies one thread from the executor’s pool, therefore, the pool should have more threads than the number of the configured WAN batch publishers to leave threads available for processing other WAN related tasks such as asynchronously processing the responses of the WAN target clusters. Since the threads WAN executor uses are borrowed from a global thread pool to reduce the number of threads in use, the threads running WAN tasks do not have any specific name.

The last two elements that you might want to change are acknowledge-type and max-concurrent-invocations. Changing these elements allow you to get a greater throughput at the expense of event ordering. This means that these elements may only be changed if your application can tolerate WAN events to be received out-of-order. For instance, if you are updating or removing the existing map or cache entries, an out-of-order WAN event delivery would mean that the event for the entry removal or update might be processed by the target cluster before the event is received to create that entry. This does not causes exceptions but it causes the clusters to fall out-of-sync. In these cases, you most probably will not be able to use these elements. On the other hand, if you are only creating new, immutable entries (which are then removed by the expiration mechanism), you can use these elements to achieve a greater throughput.

The acknowledge-type element controls at which time the target cluster will send a response for the received WAN event batch. The default value is ACK_ON_OPERATION_COMPLETE which will ensure that all events are processed before the response is sent to the source cluster. The value ACK_ON_RECEIPT instructs the target cluster to send a response as soon as it has received the WAN event batch but before it has been processed. This has two implications. One is that events can now be processed out-of-order (see the previous paragraph) and the other is that the exceptions thrown on processing the WAN event batch will not be received by the source cluster and the WAN event batch will not be retried. As such, some events might get lost in case of errors and the clusters may fall out-of-sync. WAN sync can help bring those clusters in-sync. The benefit of the ACK_ON_RECEIPT value is that now the source cluster can send a new batch sooner, without waiting for the previous batch to be processed fully.

The max-concurrent-invocations element controls the maximum number of WAN event batches being sent to the target cluster concurrently. Setting this element to anything less than 2 will only allow a single batch of events to be sent to each target endpoint and will maintain causality of events for a single partition (events are not received out-of-order). Setting this element to 2 or higher will allow multiple batches of WAN events to be sent to each target endpoint. Since this allows reordering of batches due to the network conditions, causality and ordering of events for a single partition is lost and batches for a single partition are now sent randomly to any available target endpoint. This, however, does present a faster WAN replication for certain scenarios such as replicating immutable, independent map entries which are only added once and where ordering, when these entries are added, is not necessary. Keep in mind that if you set this element to a value which is less than the target endpoint count, you will lose performance as not all target endpoints will be used at any point in time to process the WAN event batches. So, for instance, if you have a target cluster with 3 members (target endpoints) and you want to use this element, it makes sense to set it to a value equal to or greater than 3. Otherwise, you can simply disable it by setting it to less than 2 in which case WAN will use the default replication strategy and adapt to the target endpoint count while maintaining causality.

An example WAN replication configuration using the default values of the aforementioned elements is shown below.

Finally, as we’ve mentioned, the exact values which will give you the optimal performance depend on your environment and use case. Please benchmark and try out different values to find out the right values for you.

When using WAN Replication with Discovery SPI, you can set the period in seconds in which WAN tries to discover new target endpoints and reestablish connections to failed endpoints using the discovery-period-seconds property. The default value is 10 seconds.

When using WAN Replication with Discovery SPI, you can set the maximum number of endpoints that WAN connects to at any point using the max-target-endpoints property. This element has no effect when static endpoint addresses are defined using target-endpoints. Default is Integer.MAX_VALUE.

When using WAN Replication with Discovery SPI, you can set whether the WAN connection manager should connect to the endpoint on the private address returned by the Discovery SPI using the use-endpoint-private-address property. By default, this element is false which means the WAN connection manager always uses the public address.

The replication of eviction events can be useful for maintaining stronger data consistency between clusters. However, you should use it with caution; source clusters with eviction enabled should replicate to target clusters which do not have eviction enabled. If both source and target clusters have eviction enabled, due to eviction being a largely local operation, it is possible for a target cluster to evict data that has not been evicted on the source cluster. When replication is used for data consistency on a failover deployment, it is important that eviction is re-enabled on the standby cluster when it becomes active.

While enabling this functionality can help to maintain stronger data consistency over WAN between clusters, it is important to note that this does not guarantee full data consistency. You can enable the replication of IMap and ICache evictions to the other clusters by setting the hazelcast.wan.replicate.imap/icache.evictions property to true, as shown below.

You can find the specifications of these properties in System Properties.

The default behavior for the static endpoint discovery strategy is to perform health checks on the endpoints. There is no additional configuration required beyond defining the static endpoints as our default values cover the most use-cases. However, you can configure the cluster using the following properties for your specific needs.