org.opends.server.replication.service

Class ReplicationDomain

java.lang.Object

org.opends.server.replication.service.ReplicationDomain

Direct Known Subclasses:

LDAPReplicationDomain

public abstract class ReplicationDomain
extends Object

This class should be used as a base for Replication implementations.

It is intended that developer in need of a replication mechanism subclass this class with their own implementation.

The startup phase of the ReplicationDomain subclass, should read the list of replication servers from the configuration, instantiate a ServerState then start the publish service by calling startPublishService(). At this point it can start calling the #publish(UpdateMsg) method if needed.

When the startup phase reach the point when the subclass is ready to handle updates the Replication Domain implementation should call the startListenService() method. At this point a Listener thread is created on the Replication Service and which can start receiving updates.

When updates are received the Replication Service calls the dispatchUpdateForReplay(UpdateMsg) method. ReplicationDomain implementation should implement the appropriate code for replaying the update on the local repository. When fully done the subclass must call the processUpdateAfterReplay(UpdateMsg) method. This allows to process the update asynchronously if necessary.

To propagate changes to other replica, a ReplicationDomain implementation must use the #publish(UpdateMsg) method.

If the Full Initialization process is needed then implementation for importBackend(InputStream) and exportBackend(OutputStream) must be provided.

Full Initialization of a replica can be triggered by LDAP clients by creating InitializeTasks or InitializeTargetTask. Full initialization can also be triggered from the ReplicationDomain implementation using methods #initializeRemote(int, Task) or #initializeFromRemote(int, Task).

At shutdown time, the disableService() method should be called to cleanly stop the replication service.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	ReplicationDomain.ImportExportContext This class contains the context related to an import or export launched on the domain.

Field Summary

Fields

Modifier and Type	Field and Description
protected ReplicationBroker	broker The ReplicationBroker that is used by this ReplicationDomain to connect to the ReplicationService.
protected ReplicationDomainCfg	config The configuration of the replication domain.
protected long	generationId The generationId for this replication domain.

Constructor Summary

Constructors

Constructor and Description
ReplicationDomain(ReplicationContext replicationContext, ReplicationDomainCfg config, long generationId) Creates a ReplicationDomain with the provided parameters.
ReplicationDomain(ReplicationContext replicationContext, ReplicationDomainCfg config, long generationId, ServerState serverState) Creates a ReplicationDomain with the provided parameters.

Method Summary

Modifier and Type	Method and Description
void	addAdditionalMonitoring(MeterRegistryHolder registry) Subclasses should use this method to add additional monitoring information in the ReplicationDomain.
protected void	changeConfig(ReplicationDomainCfg config) Change some ReplicationDomain parameters.
void	changeConfig(Set<String> includeAttributes, Set<String> includeAttributesForDeletes) Applies a configuration change to the attributes which should be included in the ECL.
abstract long	countEntries() This method should return the total number of objects in the replicated domain.
ReplicaId	decodeTarget(String targetString) Verifies that the given string represents a valid source from which this server can be initialized.
void	disableService() Temporarily disable the Replication Service.
abstract void	dispatchUpdateForReplay(UpdateMsg updateMsg) This method ensures this UpdateMsg received from remote replication entities will be replayed, by dispatching it to the replay threads.
void	enableService() Restart the Replication service after a disableService().
protected abstract void	exportBackend(OutputStream output) This method should trigger an export of the replicated data.
Dn	getBaseDN() Returns the base DN of this ReplicationDomain.
CSNGenerator	getCsnGenerator() Returns the CSNGenerator that will be used to generate CSN for this domain.
Set<String>	getEclIncludes() Get the attributes to include in each change for the ECL.
Set<String>	getEclIncludesForDeletes() Get the attributes to include in each delete change for the ECL.
long	getGenerationID() This method should return the generationID to use for this ReplicationDomain.
GroupId	getGroupId() Gets the group id for this domain.
HealthStatus	getHealthStatus(long delayThreshold) Returns the health status based on the current replication delay.
protected ReplicationDomain.ImportExportContext	getImportExportContext() Returns the Import/Export context associated to this ReplicationDomain.
CSN	getLastLocalChange() Returns the CSN of the last Change that was fully processed by this ReplicationDomain.
Entry	getMonitorEntry() Returns the monitor entry for this replication domain.
Set<String>	getRefUrls() Gets the referrals URLs this domain publishes.
ReplicaId	getReplicaId() Get the replica ID which identifies this Replication Domain inside the Replication Service.
Map<ReplicaId,DSInfo>	getReplicaInfos() Gets the info for Replicas in the topology (except us).
Map<ReplicaId,ServerState>	getReplicaStates() Gets the States of all the Replicas currently in the Topology.
ReplicationContext	getReplicationContext() Returns the replication context.
ReplicationServerId	getReplicationServerId() Gets the server ID of the Replication Server to which the domain is currently connected.
int	getReplicationServerPort() Get the port of the replicationServer to which this domain is currently connected.
List<RSInfo>	getRsInfos() Gets the info for RSs in the topology (except the one we are connected to).
ServerState	getServerState() Get the ServerState maintained by the Concrete class.
ServerStatus	getStatus() Gets the status for this domain.
boolean	hasConnectionError() Check if the domain has a connection error.
boolean	ieRunning() Returns a boolean indicating if an import or export is currently processed.
protected abstract void	importBackend(InputStream input) This method should trigger an import of the replicated data.
void	initializeFromRemote(ReplicaId source, Task initTask) Initializes asynchronously this domain from a remote source server.
protected void	initializeRemote(ReplicaId replicaToInitialize, ReplicaId replicaRunningTheTask, Task initTask, int initWindow) Process the initialization of some other server or servers in the topology specified by the target argument when this initialization specifying the server that requests the initialization.
void	initializeRemote(ReplicaId target, Task initTask) Initializes a remote server from this server.
boolean	isConnected() Check if the domain is connected to a ReplicationServer.
protected boolean	isListenerShuttingDown() Returns true if the listener thread is shutting down or has shutdown.
protected void	processUpdateAfterReplay(UpdateMsg msg) This method must be called after each call to dispatchUpdateForReplay(UpdateMsg) when the processing of the update is completed.
void	publish(ReplicationMsg msg) Publish an UpdateMsg to the Replication Service.
void	publishHeartbeatMsg() Publishes a heartbeat message if all pending changes for current replica have been sent out.
void	publishReplicaOfflineMsg() Publishes a replica offline message if all pending changes for current replica have been sent out.
protected byte[]	receiveEntryBytes() Receives bytes related to an entry in the context of an import to initialize the domain (called by ReplLDIFInputStream).
protected void	recreateRemoteReplicasFromState() Rebuilds Remote DS lists after a reset of the domain state, either at startup or domain re-initialization.
void	resetGenerationId(Long generationIdNewValue) Reset the generationId of this domain in the whole topology.
void	sessionInitiated(ServerStatus initStatus, ServerState rsState) Set the initial status of the domain and perform necessary initializations.
boolean	setEclIncludes(ReplicaId replicaId, Set<String> includeAttributes, Set<String> includeAttributesForDeletes) Set the attributes configured on a server to be included in the ECL.
void	setGenerationID(long generationId) Sets the generationId for this replication domain.
protected void	signalNewStatus(StatusMachineEvent event) Sets the status to a new value depending of the passed status machine event.
void	startListenService() Starts the receiver side of the Replication Service.
void	startPublishService() Start the publish mechanism of the Replication Service.
String	toString()
protected void	updateState() Update the server state if necessary.

Methods inherited from class java.lang.Object

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

Field Detail

config

protected volatile ReplicationDomainCfg config

The configuration of the replication domain.

broker

protected ReplicationBroker broker

The ReplicationBroker that is used by this ReplicationDomain to connect to the ReplicationService.

generationId

protected volatile long generationId

The generationId for this replication domain. It is made of a hash of the 1000 first entries for this domain.

Constructor Detail

ReplicationDomain

public ReplicationDomain(ReplicationContext replicationContext,
                         ReplicationDomainCfg config,
                         long generationId)

Creates a ReplicationDomain with the provided parameters.

Parameters:

replicationContext - The replication context

config - The configuration object for this ReplicationDomain

generationId - the generation of this ReplicationDomain

ReplicationDomain

public ReplicationDomain(ReplicationContext replicationContext,
                         ReplicationDomainCfg config,
                         long generationId,
                         ServerState serverState)

Creates a ReplicationDomain with the provided parameters. (for unit test purpose only)

Parameters:

replicationContext - The replication context

config - The configuration object for this ReplicationDomain

generationId - the generation of this ReplicationDomain

serverState - The serverState to use

Method Detail

getHealthStatus

public HealthStatus getHealthStatus(long delayThreshold)

Returns the health status based on the current replication delay. If the delay is above the provided threshold, the replication domain is considered not healthy.

Parameters:

delayThreshold - The maximum replication delay in milliseconds for considering this replication domain healthy.

Returns:

This replication domain health status.

getCsnGenerator

public CSNGenerator getCsnGenerator()

Returns the CSNGenerator that will be used to generate CSN for this domain.

Returns:

The CSNGenerator that will be used to generate CSN for this domain.

sessionInitiated

public void sessionInitiated(ServerStatus initStatus,
                             ServerState rsState)

Set the initial status of the domain and perform necessary initializations. This method will be called by the Broker each time the ReplicationBroker establish a new session to a Replication Server. Implementations may override this method when they need to perform additional computing after session establishment. The default implementation should be sufficient for ReplicationDomains that don't need to perform additional computing.

Parameters:

initStatus - The status to enter the state machine with.

rsState - The ServerState of the ReplicationServer with which the session was established.

getStatus

public ServerStatus getStatus()

Gets the status for this domain.

Returns:

The status for this domain.

getBaseDN

public Dn getBaseDN()

Returns the base DN of this ReplicationDomain. All Replication Domain using this baseDN will be connected through the Replication Service.

Returns:

The base DN of this ReplicationDomain

getReplicaId

public ReplicaId getReplicaId()

Get the replica ID which identifies this Replication Domain inside the Replication Service.

Returns:

The replica ID.

getReplicationContext

public ReplicationContext getReplicationContext()

Returns the replication context.

Returns:

the replication context

getGroupId

public GroupId getGroupId()

Gets the group id for this domain.

Returns:

The group id for this domain.

getRefUrls

public Set<String> getRefUrls()

Gets the referrals URLs this domain publishes. Referrals urls to be published to other servers of the topology.

TODO: fill that with all currently opened urls if no urls configured

Returns:

The referrals URLs this domain publishes.

getReplicaInfos

public Map<ReplicaId,DSInfo> getReplicaInfos()

Gets the info for Replicas in the topology (except us).

Returns:

The info for Replicas in the topology (except us)

getReplicaStates

public Map<ReplicaId,ServerState> getReplicaStates()

Gets the States of all the Replicas currently in the Topology.

When this method is called, a Monitoring message will be sent to the Replication Server to which this domain is currently connected so that it computes a table containing information about all Directory Servers in the topology. This Computation involves communications will all the servers currently connected.

Returns:

The States of all Replicas in the topology (except us)

getRsInfos

public List<RSInfo> getRsInfos()

Gets the info for RSs in the topology (except the one we are connected to).

Returns:

The info for RSs in the topology (except the one we are connected to)

getReplicationServerId

public ReplicationServerId getReplicationServerId()

Gets the server ID of the Replication Server to which the domain is currently connected.

Returns:

The server ID of the Replication Server to which the domain is currently connected.

recreateRemoteReplicasFromState

protected void recreateRemoteReplicasFromState()

Rebuilds Remote DS lists after a reset of the domain state, either at startup or domain re-initialization.

updateState

protected void updateState()

Update the server state if necessary.

decodeTarget

public ReplicaId decodeTarget(String targetString)
                       throws LdapException

Verifies that the given string represents a valid source from which this server can be initialized.

Parameters:

targetString - The string representing the source

Returns:

The source as a integer value

Throws:

LdapException - if the string is not valid

initializeRemote

public void initializeRemote(ReplicaId target,
                             Task initTask)
                      throws LdapException

Initializes a remote server from this server.

The exportBackend(OutputStream) will therefore be called on this server, and the importBackend(InputStream) will be called on the remote server.

The InputStream and OutputStream given as a parameter to those methods will be connected through the replication protocol.

Parameters:

target - The server-id of the server that should be initialized. The target can be discovered using the getReplicaInfos() method.

initTask - The task that triggers this initialization and that should be updated with its progress.

Throws:

LdapException - If it was not possible to publish the Initialization message to the Topology.

initializeRemote

protected void initializeRemote(ReplicaId replicaToInitialize,
                                ReplicaId replicaRunningTheTask,
                                Task initTask,
                                int initWindow)
                         throws LdapException

Process the initialization of some other server or servers in the topology specified by the target argument when this initialization specifying the server that requests the initialization.

Parameters:

replicaToInitialize - The target replica that should be initialized.

replicaRunningTheTask - The replica that initiated the export. It can be the replica id of this server, or the replica id of a remote replica.

initTask - The task in this server that triggers this initialization and that should be updated with its progress. Null when the export is done following a request coming from a remote server (task is remote).

initWindow - The value of the initialization window for flow control between the importer and the exporter.

Throws:

LdapException - When an error occurs. No exception raised means success.

getServerState

public ServerState getServerState()

Get the ServerState maintained by the Concrete class.

Returns:

the ServerState maintained by the Concrete class.

receiveEntryBytes

protected byte[] receiveEntryBytes()

Receives bytes related to an entry in the context of an import to initialize the domain (called by ReplLDIFInputStream).

Returns:

The bytes. Null when the Done or Err message has been received

initializeFromRemote

public void initializeFromRemote(ReplicaId source,
                                 Task initTask)
                          throws LdapException

Initializes asynchronously this domain from a remote source server. Before returning from this call, for the provided task :

1. the progressing counters are updated during the initialization using setTotal() and setLeft().
2. the end of the initialization using updateTaskCompletionState().

When this method is called, a request for initialization is sent to the remote source server requesting initialization.

Parameters:

source - The server-id of the source from which to initialize. The source can be discovered using the getReplicaInfos() method.

initTask - The task that launched the initialization and should be updated of its progress.

Throws:

LdapException - If it was not possible to publish the Initialization message to the Topology. The task state is updated.

signalNewStatus

protected void signalNewStatus(StatusMachineEvent event)

Sets the status to a new value depending of the passed status machine event.

Parameters:

event - The event that may make the status be changed

ieRunning

public boolean ieRunning()

Returns a boolean indicating if an import or export is currently processed.

Returns:

The status

resetGenerationId

public void resetGenerationId(Long generationIdNewValue)
                       throws LdapException

Reset the generationId of this domain in the whole topology. A message is sent to the Replication Servers for them to reset their change dbs.

Parameters:

generationIdNewValue - The new value of the generation Id.

Throws:

LdapException - When an error occurs

isConnected

public boolean isConnected()

Check if the domain is connected to a ReplicationServer.

Returns:

true if the server is connected, false if not.

hasConnectionError

public boolean hasConnectionError()

Check if the domain has a connection error. A Connection error happens when the broker could not be created or when the broker could not find any ReplicationServer to connect to.

Returns:

true if the domain has a connection error.

getReplicationServerPort

public int getReplicationServerPort()

Get the port of the replicationServer to which this domain is currently connected.

Returns:

the replication server port to which this domain is currently connected.

getMonitorEntry

public Entry getMonitorEntry()

Returns the monitor entry for this replication domain.

Returns:

the monitor entry for this replication domain

startPublishService

public void startPublishService()

Start the publish mechanism of the Replication Service. After this method has been called, the publish service can be used by calling the #publish(UpdateMsg) method.

startListenService

public void startListenService()

Starts the receiver side of the Replication Service.

After this method has been called, the Replication Service will start calling the dispatchUpdateForReplay(UpdateMsg).

This method must be called once and must be called after the startPublishService().

disableService

public void disableService()

Temporarily disable the Replication Service. The Replication Service can be enabled again using enableService().

It can be useful to disable the Replication Service when the repository where the replicated information is stored becomes temporarily unavailable and replicated updates can therefore not be replayed during a while. This method is not MT safe.

isListenerShuttingDown

protected final boolean isListenerShuttingDown()

Returns true if the listener thread is shutting down or has shutdown.

Returns:

true if the listener thread is shutting down or has shutdown.

enableService

public void enableService()

Restart the Replication service after a disableService().

The Replication Service will restart from the point indicated by the ServerState that was given as a parameter to the startPublishService() at startup time.

If some data have changed in the repository during the period of time when the Replication Service was disabled, this ServerState should therefore be updated by the Replication Domain subclass before calling this method. This method is not MT safe.

changeConfig

protected void changeConfig(ReplicationDomainCfg config)

Change some ReplicationDomain parameters.

Parameters:

config - The new configuration that this domain should now use.

changeConfig

public void changeConfig(Set<String> includeAttributes,
                         Set<String> includeAttributesForDeletes)

Applies a configuration change to the attributes which should be included in the ECL.

Parameters:

includeAttributes - attributes to be included with all change records.

includeAttributesForDeletes - additional attributes to be included with delete change records.

exportBackend

protected abstract void exportBackend(OutputStream output)
                               throws LdapException

This method should trigger an export of the replicated data. to the provided outputStream. When finished the outputStream should be flushed and closed.

Parameters:

output - The OutputStream where the export should be produced.

Throws:

LdapException - When needed.

importBackend

protected abstract void importBackend(InputStream input)
                               throws LdapException

This method should trigger an import of the replicated data.

Parameters:

input - The InputStream from which the import should be reading entries.

Throws:

LdapException - When needed.

countEntries

public abstract long countEntries()
                           throws LdapException

This method should return the total number of objects in the replicated domain. This count will be used for reporting.

Returns:

The number of objects in the replication domain.

Throws:

LdapException - when needed.

dispatchUpdateForReplay

public abstract void dispatchUpdateForReplay(UpdateMsg updateMsg)

This method ensures this UpdateMsg received from remote replication entities will be replayed, by dispatching it to the replay threads.

Parameters:

updateMsg - The UpdateMsg to replay.

processUpdateAfterReplay

protected final void processUpdateAfterReplay(UpdateMsg msg)

This method must be called after each call to dispatchUpdateForReplay(UpdateMsg) when the processing of the update is completed.

It is useful for implementation needing to process the update in an asynchronous way or using several threads, but must be called even by implementation doing it in a synchronous, single-threaded way.

Parameters:

msg - The UpdateMsg whose processing was completed.

publish

public void publish(ReplicationMsg msg)

Publish an UpdateMsg to the Replication Service.

The Replication Service will handle the delivery of this UpdateMsg to all the participants of this Replication Domain. These members will be receive this UpdateMsg through a call of the dispatchUpdateForReplay(UpdateMsg) message.

Parameters:

msg - The UpdateMsg that should be published.

publishHeartbeatMsg

public void publishHeartbeatMsg()

Publishes a heartbeat message if all pending changes for current replica have been sent out.

publishReplicaOfflineMsg

public void publishReplicaOfflineMsg()

Publishes a replica offline message if all pending changes for current replica have been sent out.

getGenerationID

public long getGenerationID()

This method should return the generationID to use for this ReplicationDomain. This method can be called at any time after the ReplicationDomain has been started.

Returns:

The GenerationID.

setGenerationID

public void setGenerationID(long generationId)

Sets the generationId for this replication domain.

Parameters:

generationId - the generationId to set

addAdditionalMonitoring

public void addAdditionalMonitoring(MeterRegistryHolder registry)

Subclasses should use this method to add additional monitoring information in the ReplicationDomain.

Parameters:

registry - where to additional monitoring attributes

getImportExportContext

protected ReplicationDomain.ImportExportContext getImportExportContext()

Returns the Import/Export context associated to this ReplicationDomain.

Returns:

the Import/Export context associated to this ReplicationDomain

setEclIncludes

public boolean setEclIncludes(ReplicaId replicaId,
                              Set<String> includeAttributes,
                              Set<String> includeAttributesForDeletes)

Set the attributes configured on a server to be included in the ECL.

Parameters:

replicaId - Server where these attributes are configured.

includeAttributes - Attributes to be included with all change records, may include wild-cards.

includeAttributesForDeletes - Additional attributes to be included with delete change records, may include wild-cards.

Returns:

true if the set of attributes was modified.

getEclIncludes

public Set<String> getEclIncludes()

Get the attributes to include in each change for the ECL.

Returns:

The attributes to include in each change for the ECL.

getEclIncludesForDeletes

public Set<String> getEclIncludesForDeletes()

Get the attributes to include in each delete change for the ECL.

Returns:

The attributes to include in each delete change for the ECL.

getLastLocalChange

public CSN getLastLocalChange()

Returns the CSN of the last Change that was fully processed by this ReplicationDomain.

Returns:

The CSN of the last Change that was fully processed by this ReplicationDomain.

toString

public String toString()

Overrides:

toString in class Object