com.hazelcast.enterprise.wan.replication

Class AbstractWanPublisher

java.lang.Object

com.hazelcast.enterprise.wan.replication.AbstractWanPublisher

All Implemented Interfaces:

WanEventQueueMigrationListener, WanReplicationEndpoint, com.hazelcast.wan.WanReplicationPublisher

Direct Known Subclasses:

AbstractWanReplication

public abstract class AbstractWanPublisher
extends Object
implements com.hazelcast.wan.WanReplicationPublisher, WanReplicationEndpoint, WanEventQueueMigrationListener

Abstract WAN event publisher implementation. This implementation prepares the WAN events from the WAN queues and from WAN sync events for sending to the WAN endpoints.

Field Summary

Fields

Modifier and Type	Field and Description
protected WanConfigurationContext	configurationContext
protected PollSynchronizerPublisherQueueContainer	eventQueueContainer
protected long	lastQueueFullLogTimeMs
protected String	localGroupName
protected com.hazelcast.logging.ILogger	logger
protected com.hazelcast.instance.Node	node
protected int	queueCapacity
protected com.hazelcast.config.WANQueueFullBehavior	queueFullBehavior
protected int	queueLoggerTimePeriodMs
protected boolean	running
protected com.hazelcast.config.WanPublisherState	state
protected WanPublisherSyncSupport	syncSupport
protected com.hazelcast.enterprise.wan.replication.WanElementCounter	wanCounter
protected String	wanPublisherId
protected String	wanReplicationName

Constructor Summary

Constructors

Constructor and Description
AbstractWanPublisher()

Method Summary

Modifier and Type	Method and Description
void	addCacheQueue(String name, int partitionId, WanReplicationEventQueue eventQueue) Publishes all events from the eventQueue to the WAN queue for the given partition ID and cache name.
void	addMapQueue(String name, int partitionId, WanReplicationEventQueue eventQueue) Publishes all events from the eventQueue to the WAN queue for the given partition ID and map name.
protected void	afterShutdown()
void	checkWanReplicationQueues()
void	clearQueues() Clears the WAN queues and recalculates the WAN queue sizes.
void	collectAllServiceNamespaces(com.hazelcast.spi.PartitionReplicationEvent event, Set<com.hazelcast.spi.ServiceNamespace> namespaces) Collect the namespaces of all queues that should be replicated by the replication event.
void	collectReplicationData(String wanReplicationName, com.hazelcast.spi.PartitionReplicationEvent event, Collection<com.hazelcast.spi.ServiceNamespace> namespaces, EWRMigrationContainer migrationDataContainer) Collect all replication data for the specific replication event and collection of namespaces being replicated.
protected abstract WanPublisherSyncSupport	createWanSyncSupport() Creates the instance implementing WAN sync features.
void	decrementCounter(int delta, boolean isPrimary) Decrements the primary or backup counter by the given delta.
void	destroyMapData(String mapName) Releases all resources for the map with the given mapName.
int	getCurrentBackupElementCount() Returns the count for the number of events currently in the WAN map/cache partition queues for partitions for which this node is a backup replica.
int	getCurrentElementCount() Returns the count for the number of events currently in the WAN map/cache partition queues for partitions owned by this node.
protected int	getPartitionId(Object key) Returns the partition ID for the partition owning the key
PublisherQueueContainer	getPublisherQueueContainer() Returns the WAN event queue container for this publisher.
com.hazelcast.monitor.LocalWanPublisherStats	getStats() Gathers statistics of related WanReplicationEndpoint instance.
void	init(com.hazelcast.instance.Node node, com.hazelcast.config.WanReplicationConfig wanReplicationConfig, com.hazelcast.config.WanPublisherConfig publisherConfig) Initializes the endpoint using the given arguments.
abstract boolean	isConnected() Tests is there is at least one alive connection to a configured target cluster member over WAN.
void	onMigrationCommit(int partitionId, int currentReplicaIndex, int newReplicaIndex) Indicates that migration is committing for a given partition
void	onMigrationRollback(int partitionId, int currentReplicaIndex, int newReplicaIndex) Indicates that migration is rolling back for a given partition
void	onMigrationStart(int partitionId, int currentReplicaIndex, int newReplicaIndex) Indicates that migration started for a given partition
void	onWanQueueClearedDuringMigration(int partitionId, int currentReplicaIndex, int clearedQueueDepth) Indicates that a partition's WAN event queue is cleared during migration
void	pause() Calls to this method will pause WAN event queue polling.
abstract void	publishAntiEntropyEvent(WanAntiEntropyEvent event) Publishes a WAN anti-entropy event.
void	publishReplicationEvent(String serviceName, com.hazelcast.wan.ReplicationEventObject eventObject)
void	publishReplicationEvent(com.hazelcast.wan.WanReplicationEvent wanReplicationEvent) This implementation achieves ordering of keys in the same partition by sending EWRPutOperation to the partition owner.
void	publishReplicationEventBackup(String serviceName, com.hazelcast.wan.ReplicationEventObject eventObject)
void	publishSyncEvent(WanSyncEvent event) Publishes a WAN sync event for all or a specific map and for all or some partitions.
void	putBackup(com.hazelcast.wan.WanReplicationEvent wanReplicationEvent) Puts the given wanReplicationEvent in the corresponding WAN backup queue.
void	removeBackup(com.hazelcast.wan.WanReplicationEvent wanReplicationEvent) Remove the oldest replication event from the replication queue and decrease the backup event count.
void	removeBackups(String serviceName, String objectName, int partitionId, int count) Removes a count number of events from a WAN replication queue belonging to the provided service, object and partition.
void	removeReplicationEvent(com.hazelcast.wan.WanReplicationEvent wanReplicationEvent) Updates the WAN statistics (remaining map sync events, synced partitions, queued event count, latencies, sent events) and removes the backup events on the replicas.
void	reset() Resets the publisher.
void	resume() This method re-enables WAN event queue polling for a paused or stopped WanReplicationEndpoint instance.
void	shutdown() Signals the publisher to shut down and clean up its resources.
void	stop() Calls to this method will stop WAN replication.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

wanCounter

protected final com.hazelcast.enterprise.wan.replication.WanElementCounter wanCounter

state

protected volatile com.hazelcast.config.WanPublisherState state

running

protected volatile boolean running

lastQueueFullLogTimeMs

protected volatile long lastQueueFullLogTimeMs

queueCapacity

protected int queueCapacity

queueLoggerTimePeriodMs

protected int queueLoggerTimePeriodMs

node

protected com.hazelcast.instance.Node node

logger

protected com.hazelcast.logging.ILogger logger

localGroupName

protected String localGroupName

wanPublisherId

protected String wanPublisherId

wanReplicationName

protected String wanReplicationName

queueFullBehavior

protected com.hazelcast.config.WANQueueFullBehavior queueFullBehavior

configurationContext

protected WanConfigurationContext configurationContext

eventQueueContainer

protected PollSynchronizerPublisherQueueContainer eventQueueContainer

syncSupport

protected WanPublisherSyncSupport syncSupport

Constructor Detail

AbstractWanPublisher

public AbstractWanPublisher()

Method Detail

init

public void init(com.hazelcast.instance.Node node,
                 com.hazelcast.config.WanReplicationConfig wanReplicationConfig,
                 com.hazelcast.config.WanPublisherConfig publisherConfig)

Description copied from interface: WanReplicationEndpoint

Initializes the endpoint using the given arguments.

Specified by:

init in interface WanReplicationEndpoint

Parameters:

node - the current node that tries to connect

wanReplicationConfig - the replication config

publisherConfig - this endpoint will be initialized using this WanPublisherConfig instance

getPartitionId

protected int getPartitionId(Object key)

Returns the partition ID for the partition owning the key

publishReplicationEvent

public void publishReplicationEvent(String serviceName,
                                    com.hazelcast.wan.ReplicationEventObject eventObject)

Specified by:

publishReplicationEvent in interface com.hazelcast.wan.WanReplicationPublisher

publishReplicationEventBackup

public void publishReplicationEventBackup(String serviceName,
                                          com.hazelcast.wan.ReplicationEventObject eventObject)

Specified by:

publishReplicationEventBackup in interface com.hazelcast.wan.WanReplicationPublisher

removeBackups

public void removeBackups(String serviceName,
                          String objectName,
                          int partitionId,
                          int count)

Removes a count number of events from a WAN replication queue belonging to the provided service, object and partition.

Parameters:

serviceName - the service name for the WAN replication queue

objectName - the object name for the WAN replication queue

partitionId - the partition ID of the WAN replication queue

count - the number of events to remove from the queue

removeReplicationEvent

public void removeReplicationEvent(com.hazelcast.wan.WanReplicationEvent wanReplicationEvent)

Updates the WAN statistics (remaining map sync events, synced partitions, queued event count, latencies, sent events) and removes the backup events on the replicas.

Parameters:

wanReplicationEvent - the completed replication event

Throws:

com.hazelcast.nio.serialization.HazelcastSerializationException - when the event fails to serialization

removeBackup

public void removeBackup(com.hazelcast.wan.WanReplicationEvent wanReplicationEvent)

Description copied from interface: WanReplicationEndpoint

Remove the oldest replication event from the replication queue and decrease the backup event count. The replication queue chosen has the same map/cache name and partition ID as the provided wanReplicationEvent.

Specified by:

removeBackup in interface WanReplicationEndpoint

Parameters:

wanReplicationEvent - the completed wan event

putBackup

public void putBackup(com.hazelcast.wan.WanReplicationEvent wanReplicationEvent)

Description copied from interface: WanReplicationEndpoint

Puts the given wanReplicationEvent in the corresponding WAN backup queue.

Specified by:

putBackup in interface WanReplicationEndpoint

Parameters:

wanReplicationEvent - the WAN event

getPublisherQueueContainer

public PublisherQueueContainer getPublisherQueueContainer()

Description copied from interface: WanReplicationEndpoint

Returns the WAN event queue container for this publisher.

Specified by:

getPublisherQueueContainer in interface WanReplicationEndpoint

addMapQueue

public void addMapQueue(String name,
                        int partitionId,
                        WanReplicationEventQueue eventQueue)

Description copied from interface: WanReplicationEndpoint

Publishes all events from the eventQueue to the WAN queue for the given partition ID and map name.

Specified by:

addMapQueue in interface WanReplicationEndpoint

Parameters:

name - the name of the map to which to publish to

partitionId - the partition to which to publish to

eventQueue - the event queue from which events are published

addCacheQueue

public void addCacheQueue(String name,
                          int partitionId,
                          WanReplicationEventQueue eventQueue)

Description copied from interface: WanReplicationEndpoint

Publishes all events from the eventQueue to the WAN queue for the given partition ID and cache name.

Specified by:

addCacheQueue in interface WanReplicationEndpoint

Parameters:

name - the name of the cache to which to publish to

partitionId - the partition to which to publish to

eventQueue - the event queue from which events are published

shutdown

public final void shutdown()

Description copied from interface: WanReplicationEndpoint

Signals the publisher to shut down and clean up its resources. The method does not necessarily block until the endpoint has shut down.

Specified by:

shutdown in interface WanReplicationEndpoint

afterShutdown

protected void afterShutdown()

pause

public void pause()

Description copied from interface: WanReplicationEndpoint

Calls to this method will pause WAN event queue polling. Effectively, pauses WAN replication for its WanReplicationEndpoint instance.

WAN events will still be offered to WAN replication queues but they won't be polled. This means that the queues might eventually fill up and start dropping events.

Calling this method on already paused WanReplicationEndpoint instances will have no effect.

There is no synchronization with the thread polling the WAN queues and trasmitting the events to the target cluster. This means that the queues may be polled even after this method returns.

Specified by:

pause in interface WanReplicationEndpoint

See Also:

WanReplicationEndpoint.resume()

stop

public void stop()

Description copied from interface: WanReplicationEndpoint

Calls to this method will stop WAN replication. In addition to not polling events as in the WanReplicationEndpoint.pause() method, an endpoint which is stopped will not enqueue events. This method will not clear the WAN queues, though. This means that once this method returns, there might still be some WAN events enqueued but these events will not be replicated until the publisher is resumed.

Calling this method on already stopped WanReplicationEndpoint instances will have no effect.

Specified by:

stop in interface WanReplicationEndpoint

See Also:

WanReplicationEndpoint.resume()

resume

public void resume()

Description copied from interface: WanReplicationEndpoint

This method re-enables WAN event queue polling for a paused or stopped WanReplicationEndpoint instance.

Calling this method on already running WanReplicationEndpoint instances will have no effect.

Specified by:

resume in interface WanReplicationEndpoint

See Also:

WanReplicationEndpoint.pause(), WanReplicationEndpoint.stop()

getStats

public com.hazelcast.monitor.LocalWanPublisherStats getStats()

Description copied from interface: WanReplicationEndpoint

Gathers statistics of related WanReplicationEndpoint instance. This method will always return the same instance.

Specified by:

getStats in interface WanReplicationEndpoint

Returns:

LocalWanPublisherStats

isConnected

public abstract boolean isConnected()

Tests is there is at least one alive connection to a configured target cluster member over WAN.

Returns:

true if there is at least one alive connection, false otherwise

checkWanReplicationQueues

public void checkWanReplicationQueues()

Specified by:

checkWanReplicationQueues in interface com.hazelcast.wan.WanReplicationPublisher

publishReplicationEvent

public void publishReplicationEvent(com.hazelcast.wan.WanReplicationEvent wanReplicationEvent)

This implementation achieves ordering of keys in the same partition by sending EWRPutOperation to the partition owner. The partition ID is equal to the partition for the key contained in the wanReplicationEvent.

Specified by:

publishReplicationEvent in interface com.hazelcast.wan.WanReplicationPublisher

Parameters:

wanReplicationEvent - the WAN event to publish

publishSyncEvent

public void publishSyncEvent(WanSyncEvent event)

Description copied from interface: WanReplicationEndpoint

Publishes a WAN sync event for all or a specific map and for all or some partitions.

Specified by:

publishSyncEvent in interface WanReplicationEndpoint

Parameters:

event - the WAN sync request

publishAntiEntropyEvent

public abstract void publishAntiEntropyEvent(WanAntiEntropyEvent event)

Publishes a WAN anti-entropy event. This method may also process the event or trigger processing.

Parameters:

event - the WAN anti-entropy event

destroyMapData

public void destroyMapData(String mapName)

Releases all resources for the map with the given mapName.

Parameters:

mapName - the map name

clearQueues

public void clearQueues()

Description copied from interface: WanReplicationEndpoint

Clears the WAN queues and recalculates the WAN queue sizes.

Specified by:

clearQueues in interface WanReplicationEndpoint

decrementCounter

public void decrementCounter(int delta,
                             boolean isPrimary)

Decrements the primary or backup counter by the given delta.

Parameters:

delta - the delta by which we need to decrement the counter

isPrimary - true if the primary WAN event queue counter needs to be decremented, false if the backup WAN event queue counter needs to be decremented

onMigrationStart

public void onMigrationStart(int partitionId,
                             int currentReplicaIndex,
                             int newReplicaIndex)

Description copied from interface: WanEventQueueMigrationListener

Indicates that migration started for a given partition

Specified by:

onMigrationStart in interface WanEventQueueMigrationListener

Parameters:

partitionId - the partition being migrated

currentReplicaIndex - the current replica index of the partition

newReplicaIndex - the new replica index if the partition

onMigrationCommit

public void onMigrationCommit(int partitionId,
                              int currentReplicaIndex,
                              int newReplicaIndex)

Description copied from interface: WanEventQueueMigrationListener

Indicates that migration is committing for a given partition

Specified by:

onMigrationCommit in interface WanEventQueueMigrationListener

Parameters:

partitionId - the partition being migrated

currentReplicaIndex - the current replica index of the partition

newReplicaIndex - the new replica index if the partition

onMigrationRollback

public void onMigrationRollback(int partitionId,
                                int currentReplicaIndex,
                                int newReplicaIndex)

Description copied from interface: WanEventQueueMigrationListener

Indicates that migration is rolling back for a given partition

Specified by:

onMigrationRollback in interface WanEventQueueMigrationListener

Parameters:

partitionId - the partition being migrated

currentReplicaIndex - the current replica index of the partition

newReplicaIndex - the new replica index if the partition

onWanQueueClearedDuringMigration

public void onWanQueueClearedDuringMigration(int partitionId,
                                             int currentReplicaIndex,
                                             int clearedQueueDepth)

Description copied from interface: WanEventQueueMigrationListener

Indicates that a partition's WAN event queue is cleared during migration

Specified by:

onWanQueueClearedDuringMigration in interface WanEventQueueMigrationListener

Parameters:

partitionId - the partition being migrated

currentReplicaIndex - the current replica index of the partition

clearedQueueDepth - the depth of the migrated WAN queue

reset

public void reset()

Resets the publisher. Removes all WAN events from the partition queues and resets the WAN counters.

getCurrentElementCount

public int getCurrentElementCount()

Returns the count for the number of events currently in the WAN map/cache partition queues for partitions owned by this node.

getCurrentBackupElementCount

public int getCurrentBackupElementCount()

Returns the count for the number of events currently in the WAN map/cache partition queues for partitions for which this node is a backup replica.

collectAllServiceNamespaces

public void collectAllServiceNamespaces(com.hazelcast.spi.PartitionReplicationEvent event,
                                        Set<com.hazelcast.spi.ServiceNamespace> namespaces)

Description copied from interface: WanReplicationEndpoint

Collect the namespaces of all queues that should be replicated by the replication event.

Specified by:

collectAllServiceNamespaces in interface WanReplicationEndpoint

Parameters:

event - the replication event

namespaces - the set in which namespaces should be added

collectReplicationData

public void collectReplicationData(String wanReplicationName,
                                   com.hazelcast.spi.PartitionReplicationEvent event,
                                   Collection<com.hazelcast.spi.ServiceNamespace> namespaces,
                                   EWRMigrationContainer migrationDataContainer)

Description copied from interface: WanReplicationEndpoint

Collect all replication data for the specific replication event and collection of namespaces being replicated.

Specified by:

collectReplicationData in interface WanReplicationEndpoint

Parameters:

wanReplicationName - the WAN replication name in the hazelcast configuration for this endpoint

event - the replication event

namespaces - the object namespaces which are being replicated

migrationDataContainer - the container for the migration data

createWanSyncSupport

protected abstract WanPublisherSyncSupport createWanSyncSupport()

Creates the instance implementing WAN sync features. This method must not return null.