ReplicationDomain.java revision a395dd575518d9e5280fc5d5d5ef47c61b174647
/*
* CDDL HEADER START
*
* The contents of this file are subject to the terms of the
* Common Development and Distribution License, Version 1.0 only
* (the "License"). You may not use this file except in compliance
* with the License.
*
* You can obtain a copy of the license at
* See the License for the specific language governing permissions
* and limitations under the License.
*
* When distributing Covered Code, include this CDDL HEADER in each
* file and include the License file at
* trunk/opends/resource/legal-notices/OpenDS.LICENSE. If applicable,
* add the following below this CDDL HEADER, with the fields enclosed
* by brackets "[]" replaced with your own identifying information:
* Portions Copyright [yyyy] [name of copyright owner]
*
* CDDL HEADER END
*
*
* Copyright 2006-2008 Sun Microsystems, Inc.
*/
/**
* This class implements the bulk part of the.of the Directory Server side
* of the replication code.
* It contains the root method for publishing a change,
* processing a change received from the replicationServer service,
* handle conflict resolution,
* handle protocol messages from the replicationServer.
*/
public class ReplicationDomain extends DirectoryThread
implements ConfigurationChangeListener<ReplicationDomainCfg>,
{
/**
* The fully-qualified name of this class.
*/
private static final String CLASS_NAME =
"org.opends.server.replication.plugin.ReplicationDomain";
/**
* The attribute used to mark conflicting entries.
* The value of this attribute should be the dn that this entry was
* supposed to have when it was marked as conflicting.
*/
/**
* The tracer object for the debug logger.
*/
private ReplicationMonitor monitor;
private ReplicationBroker broker;
// Thread waiting for incoming update messages for this domain and pushing
// them to the global incoming update message queue for later processing by
// replay threads.
private ListenerThread listenerThread;
// The update to replay message queue where the listener thread is going to
// push incoming update messages.
private int debugCount = 0;
private PersistentServerState state;
private int numReplayedPostOpCalled = 0;
private int maxReceiveQueue = 0;
private int maxSendQueue = 0;
private int maxReceiveDelay = 0;
private int maxSendDelay = 0;
private long generationId = -1;
private boolean generationIdSavedStatus = false;
/**
* This object is used to store the list of update currently being
* done on the local database.
* Is is usefull to make sure that the local operations are sent in a
* correct order to the replication server and that the ServerState
* is not updated too early.
*/
private PendingChanges pendingChanges;
/**
* It contain the updates that were done on other servers, transmitted
* by the replication server and that are currently replayed.
* It is usefull to make sure that dependencies between operations
* are correctly fullfilled and to to make sure that the ServerState is
* not updated too early.
*/
private RemotePendingChanges remotePendingChanges;
/**
* The time in milliseconds between heartbeats from the replication
* server. Zero means heartbeats are off.
*/
private long heartbeatInterval = 0;
short serverId;
// The context related to an import or export being processed
// Null when none is being processed.
private boolean shutdown = false;
private InternalClientConnection conn =
private boolean solveConflictFlag = true;
private boolean disabled = false;
private boolean stateSavingDisabled = false;
private int window = 100;
/**
* The isolation policy that this domain is going to use.
* This field describes the behavior of the domain when an update is
* attempted and the domain could not connect to any Replication Server.
* Possible values are accept-updates or deny-updates, but other values
* may be added in the futur.
*/
private IsolationPolicy isolationpolicy;
/**
* The DN of the configuration entry of this domain.
*/
/**
* A boolean indicating if the thread used to save the persistentServerState
* is terminated.
*/
private boolean done = false;
/**
* This class contain the context related to an import or export
* launched on the domain.
*/
private class IEContext
{
// The task that initiated the operation.
// The input stream for the import
// The target in the case of an export
// The source in the case of an import
// The total entry count expected to be processed
long entryCount = 0;
// The count for the entry not yet processed
long entryLeftCount = 0;
boolean checksumOutput = false;
// The exception raised when any
long checksumOutputValue = (long)0;
/**
* @param count The value with which to initialize the counters.
*/
throws DirectoryException
{
entryCount = total;
if (initializeTask != null)
{
if (initializeTask instanceof InitializeTask)
{
}
else if (initializeTask instanceof InitializeTargetTask)
{
}
}
}
/**
* Update the counters of the task for each entry processed during
* an import or export.
*/
public void updateCounters()
throws DirectoryException
{
if (initializeTask != null)
{
if (initializeTask instanceof InitializeTask)
{
}
else if (initializeTask instanceof InitializeTargetTask)
{
}
}
}
/**
* {@inheritDoc}
*/
{
}
}
/**
* This thread is launched when we want to export data to another server that
* has requested to be initialized with the data of our backend.
*/
private class ExportThread extends DirectoryThread
{
// Id of server that will receive updates
private short target;
/**
* Constructor for the ExportThread.
*
* @param target Id of server that will receive updates
*/
public ExportThread(short target)
{
super("Export thread");
}
/**
* Run method for this class.
*/
public void run()
{
if (debugEnabled())
{
}
try
{
} catch (DirectoryException de)
{
// An error message has been sent to the peer
// Nothing more to do locally
}
if (debugEnabled())
{
}
}
}
/**
* Creates a new ReplicationDomain using configuration from configEntry.
*
* @param configuration The configuration of this ReplicationDomain.
* @param updateToReplayQueue The queue for update messages to replay.
* @throws ConfigException In case of invalid configuration.
*/
throws ConfigException
{
// Read the configuration parameters.
/*
* Modify conflicts are solved for all suffixes but the schema suffix
* because we don't want to store extra information in the schema
* ldif files.
* This has no negative impact because the changes on schema should
* not produce conflicts.
*/
{
solveConflictFlag = false;
}
else
{
solveConflictFlag = true;
}
/*
* Create a new Persistent Server State that will be used to store
* the last ChangeNmber seen from all LDAP servers in the topology.
*/
/*
* Create a replication monitor object responsible for publishing
* monitoring information below cn=monitor.
*/
monitor = new ReplicationMonitor(this);
{
baseDN.toNormalizedString()));
}
try
{
}
catch (DirectoryException e)
{
}
/*
* create the broker object used to publish and receive changes
*/
new ReplSessionSecurity(configuration));
/*
* ChangeNumberGenerator is used to create new unique ChangeNumbers
* for each operation done on the replication domain.
*/
// listen for changes on the configuration
configuration.addChangeListener(this);
// register as an AltertGenerator
}
/**
* Returns the base DN of this ReplicationDomain.
*
* @return The base DN of this ReplicationDomain
*/
{
return baseDN;
}
/**
* Implement the handleConflictResolution phase of the deleteOperation.
*
* @param deleteOperation The deleteOperation.
* @return A SynchronizationProviderResult indicating if the operation
* can continue.
*/
{
&& (!brokerIsConnected(deleteOperation)))
{
return new SynchronizationProviderResult(false);
}
{
/*
* This is a replication operation
* Check that the modified entry has the same entryuuid
* has was in the original message.
*/
{
/*
* The changes entry is not the same entry as the one on
* the original change was performed.
* Probably the original entry was renamed and replaced with
* another entry.
* We must not let the change proceed, return a negative
* result and set the result code to NO_SUCH_OBJET.
* When the operation will return, the thread that started the
* operation will try to find the correct entry and restart a new
* operation.
*/
return new SynchronizationProviderResult(false);
}
}
else
{
// There is no replication context attached to the operation
// so this is not a replication operation.
}
return new SynchronizationProviderResult(true);
}
/**
* Implement the handleConflictResolution phase of the addOperation.
*
* @param addOperation The AddOperation.
* @return A SynchronizationProviderResult indicating if the operation
* can continue.
*/
{
if ((!addOperation.isSynchronizationOperation())
&& (!brokerIsConnected(addOperation)))
{
return new SynchronizationProviderResult(false);
}
{
/*
* If an entry with the same entry uniqueID already exist then
* this operation has already been replayed in the past.
*/
{
return new SynchronizationProviderResult(false);
}
/* The parent entry may have been renamed here since the change was done
* on the first server, and another entry have taken the former dn
* of the parent entry
*/
// root entry have no parent,
// there is no need to check for it.
{
// There is a potential of perfs improvement here
// if we could avoid the following parent entry retrieval
if (parentDnFromCtx == null)
{
// The parent does not exist with the specified unique id
// stop the operation with NO_SUCH_OBJECT and let the
// conflict resolution or the dependency resolution solve this.
return new SynchronizationProviderResult(false);
}
else
{
if ((parentDnFromEntryDn != null)
{
// parentEntry has been renamed
// replication name conflict resolution is expected to fix that
// later in the flow
return new SynchronizationProviderResult(false);
}
}
}
}
return new SynchronizationProviderResult(true);
}
/**
* Check that the broker associated to this ReplicationDomain has found
* a Replication Server and that this LDAP server is therefore able to
* process operations.
* If not set the ResultCode and the response message,
* interrupt the operation, and return false
*
* @param Operation The Operation that needs to be checked.
*
* @return true when it OK to process the Operation, false otherwise.
* When false is returned the resultCode and the reponse message
* is also set in the Operation.
*/
{
{
// this policy imply that we always accept updates.
return true;
}
{
// this isolation policy specifies that the updates are denied
// when the broker is not connected.
if (broker.isConnected())
{
return true;
}
else
{
new DirectoryException(
return false;
}
}
// we should never get there as the only possible policies are
// ACCEPT_ALL_UPDATES and REJECT_ALL_UPDATES
return true;
}
/**
* Implement the handleConflictResolution phase of the ModifyDNOperation.
*
* @param modifyDNOperation The ModifyDNOperation.
* @return A SynchronizationProviderResult indicating if the operation
* can continue.
*/
{
&& (!brokerIsConnected(modifyDNOperation)))
{
return new SynchronizationProviderResult(false);
}
{
/*
* This is a replication operation
* Check that the modified entry has the same entryuuid
* as was in the original message.
*/
{
/*
* The modified entry is not the same entry as the one on
* the original change was performed.
* Probably the original entry was renamed and replaced with
* another entry.
* We must not let the change proceed, return a negative
* result and set the result code to NO_SUCH_OBJET.
* When the operation will return, the thread that started the
* operation will try to find the correct entry and restart a new
* operation.
*/
return new SynchronizationProviderResult(false);
}
{
/*
* Also check that the current id of the
* parent is the same as when the operation was performed.
*/
if ((newParentId != null) &&
{
return new SynchronizationProviderResult(false);
}
}
}
else
{
// There is no replication context attached to the operation
// so this is not a replication operation.
{
}
}
return new SynchronizationProviderResult(true);
}
/**
* Handle the conflict resolution.
* Called by the core server after locking the entry and before
* starting the actual modification.
* @param modifyOperation the operation
* @return code indicating is operation must proceed
*/
{
&& (!brokerIsConnected(modifyOperation)))
{
return new SynchronizationProviderResult(false);
}
{
// There is no replication context attached to the operation
// so this is not a replication operation.
if (modifiedEntryUUID == null)
}
else
{
// This is a replayed operation, it is necessary to
// - check if the entry has been renamed
// - check for conflicts
if ((currentEntryUUID != null) &&
{
/*
* The current modified entry is not the same entry as the one on
* the original modification was performed.
* Probably the original entry was renamed and replaced with
* another entry.
* We must not let the modification proceed, return a negative
* result and set the result code to NO_SUCH_OBJET.
* When the operation will return, the thread that started the
* operation will try to find the correct entry and restart a new
* operation.
*/
return new SynchronizationProviderResult(false);
}
/*
* Solve the conflicts between modify operations
*/
{
}
{
/*
* This operation becomes a no-op due to conflict resolution
* stop the processing and send an OK result
*/
return new SynchronizationProviderResult(false);
}
}
return new SynchronizationProviderResult(true);
}
/**
* The preOperation phase for the add Operation.
* Its job is to generate the replication context associated to the
* operation. It is necessary to do it in this phase because contrary to
* the other operations, the entry uid is not set when the handleConflict
* phase is called.
*
* @param addOperation The Add Operation.
*/
{
}
/**
* Receives an update message from the replicationServer.
* also responsible for updating the list of pending changes
* @return the received message - null if none
*/
public UpdateMessage receive()
{
{
try
{
{
// The server is in the shutdown process
return null;
}
if (debugEnabled())
if (!(msg instanceof HeartbeatMessage))
if (msg instanceof AckMessage)
{
}
else if (msg instanceof InitializeRequestMessage)
{
// Another server requests us to provide entries
// for a total update
}
else if (msg instanceof InitializeTargetMessage)
{
// Another server is exporting its entries to us
try
{
// This must be done while we are still holding the
// broker lock because we are now going to receive a
// bunch of entries from the remote server and we
// want the import thread to catch them and
// not the ListenerThread.
}
catch(DirectoryException de)
{
// Returns an error message to notify the sender
de.getMessageObject());
}
}
else if (msg instanceof ErrorMessage)
{
{
// This is an error termination for the 2 following cases :
// - either during an export
// - or before an import really started
// For example, when we publish a request and the
// replicationServer did not find any import source.
}
else
{
/* We can receive an error message from the replication server
* in the following cases :
* - we connected with an incorrect generation id
*/
errorMsg.getDetails()));
}
}
else if (msg instanceof UpdateMessage)
{
}
}
catch (SocketTimeoutException e)
{
// just retry
}
// Test if we have received and export request message and
// if that's the case handle it now.
// This must be done outside of the portion of code protected
// by the broker lock so that we keep receiveing update
// when we are doing and export and so that a possible
// closure of the socket happening when we are publishing the
// entries to the remote can be handled by the other
// replay thread when they call this method and therefore the
// broker.receive() method.
{
// Do this work in a thread to allow replay thread continue working
}
}
return update;
}
/**
* Do the necessary processing when an UpdateMessage was received.
*
* @param update The received UpdateMessage.
*/
{
}
/**
* Do the necessary processing when an AckMessage is received.
*
* @param ack The AckMessage that was received.
*/
{
synchronized (waitingAckMsgs)
{
}
{
synchronized (update)
{
}
}
}
/**
* Check if an operation must be synchronized.
* Also update the list of pending changes and the server RUV
* @param op the operation
*/
{
{
}
// Note that a failed non-replication operation might not have a change
// number.
{
// Generate a replication message for a successful non-replication
// operation.
{
/*
* This is an operation type that we do not know about
* It should never happen.
*/
return;
}
}
{
try
{
if (op.isSynchronizationOperation())
{
}
else
{
}
}
catch (NoSuchElementException e)
{
return;
}
{
synchronized (waitingAckMsgs)
{
// Add the assured message to the list of update that are
// waiting acknowledgements
}
}
if (generationIdSavedStatus != true)
{
this.saveGenerationId(generationId);
}
}
else if (!op.isSynchronizationOperation())
{
// Remove an unsuccessful non-replication operation from the pending
// changes list.
if (curChangeNumber != null)
{
}
}
if (!op.isSynchronizationOperation())
{
}
// Wait for acknowledgement of an assured message.
{
synchronized (msg)
{
{
// TODO : should have a configurable timeout to get
// out of this loop
try
{
} catch (InterruptedException e)
{ }
}
}
}
}
/**
* get the number of updates received by the replication plugin.
*
* @return the number of updates received
*/
public int getNumRcvdUpdates()
{
if (numRcvdUpdates != null)
return numRcvdUpdates.get();
else
return 0;
}
/**
* Get the number of updates sent by the replication plugin.
*
* @return the number of updates sent
*/
public int getNumSentUpdates()
{
if (numSentUpdates != null)
return numSentUpdates.get();
else
return 0;
}
/**
* Get the number of updates in the pending list.
*
* @return The number of updates in the pending list
*/
public int getPendingUpdatesCount()
{
if (pendingChanges != null)
return pendingChanges.size();
else
return 0;
}
/**
* Increment the number of processed updates.
*/
public void incProcessedUpdates()
{
}
/**
* get the number of updates replayed by the replication.
*
* @return The number of updates replayed by the replication
*/
public int getNumProcessedUpdates()
{
if (numProcessedUpdates != null)
return numProcessedUpdates.get();
else
return 0;
}
/**
* get the number of updates replayed successfully by the replication.
*
* @return The number of updates replayed successfully
*/
public int getNumReplayedPostOpCalled()
{
return numReplayedPostOpCalled;
}
/**
* get the ServerState.
*
* @return the ServerState
*/
public ServerState getServerState()
{
return state;
}
/**
* Get the debugCount.
*
* @return Returns the debugCount.
*/
public int getDebugCount()
{
return debugCount;
}
/**
* Send an Ack message.
*
* @param changeNumber The ChangeNumber for which the ack must be sent.
*/
{
}
/**
* {@inheritDoc}
*/
public void run()
{
// Create the listener thread
while (shutdown == false)
{
try
{
synchronized (this)
{
this.wait(1000);
if (!disabled && !stateSavingDisabled )
{
// save the RUV
}
}
} catch (InterruptedException e)
{ }
}
done = true;
}
/**
* Shutdown this ReplicationDomain.
*/
public void shutdown()
{
// stop the flush thread
shutdown = true;
// Stop the listener thread
if (listenerThread != null)
{
}
synchronized (this)
{
this.notify();
}
// stop the ReplicationBroker
// Wait for the listener thread to stop
if (listenerThread != null)
// wait for completion of the persistentServerState thread.
try
{
while (!done)
{
}
} catch (InterruptedException e)
{
// stop waiting when interrupted.
}
}
/**
* Get the name of the replicationServer to which this domain is currently
* connected.
*
* @return the name of the replicationServer to which this domain
* is currently connected.
*/
public String getReplicationServer()
{
return broker.getReplicationServer();
else
return "Not connected";
}
/**
* Create and replay a synchronized Operation from an UpdateMessage.
*
* @param msg The UpdateMessage to be replayed.
*/
{
boolean done = false;
boolean dependency = false;
int retryCount = 10;
boolean firstTry = true;
// Try replay the operation, then flush (replaying) any pending operation
// whose dependency has been replayed until no more left.
do
{
try
{
{
op.setInternalOperation(true);
op.setSynchronizationOperation(true);
// Try replay the operation
{
if (op instanceof ModifyOperation)
{
if (!dependency)
{
}
} else if (op instanceof DeleteOperation)
{
if ((!dependency) && (!firstTry))
{
}
} else if (op instanceof AddOperation)
{
if (!dependency)
{
}
} else if (op instanceof ModifyDNOperationBasis)
{
if (!dependency)
{
}
} else
{
done = true; // unknown type of operation ?!
}
if (done)
{
// the update became a dummy update and the result
// of the conflict resolution phase is to do nothing.
// however we still need to push this change to the serverState
}
} else
{
done = true;
}
firstTry = false;
}
if (!done && !dependency)
{
// Continue with the next change but the servers could now become
// inconsistent.
// Let the repair tool know about this.
}
} catch (ASN1Exception e)
{
} catch (LDAPException e)
{
} catch (DataFormatException e)
{
} catch (Exception e)
{
if (changeNumber != null)
{
/*
* An Exception happened during the replay process.
* Continue with the next change but the servers will now start
* to be inconsistent.
* Let the repair tool know about this.
*/
} else
{
}
} finally
{
if (!dependency)
{
}
}
// Now replay any pending update that had a dependency and whose
// dependency has been replayed, do that until no more updates of that
// type left...
// Prepare restart of loop
done = false;
dependency = false;
changeNumber = null;
retryCount = 10;
firstTry = true;
}
/**
* This method is called when an error happens while replaying
* an operation.
* It is necessary because the postOperation does not always get
* called when error or Exceptions happen during the operation replay.
*
* @param changeNumber the ChangeNumber of the operation with error.
*/
{
}
/**
* Generate a new change number and insert it in the pending list.
*
* @param operation The operation for which the change number must be
* generated.
* @return The new change number.
*/
{
}
/**
* Find the Unique Id of the entry with the provided DN by doing a
* search of the entry and extracting its uniqueID from its attributes.
*
* @param dn The dn of the entry for which the unique Id is searched.
*
* @return The unique Id of the entry whith the provided DN.
*/
{
return null;
try
{
0, 0, false,
attrs);
{
{
if (resultEntry != null)
{
}
}
}
} catch (DirectoryException e)
{
// never happens because the filter is always valid.
}
return null;
}
/**
* find the current dn of an entry from its entry uuid.
*
* @param uuid the Entry Unique ID.
* @return The curernt dn of the entry or null if there is no entry with
* the specified uuid.
*/
{
try
{
{
{
if (resultEntry != null)
{
return resultEntry.getDN();
}
}
}
} catch (DirectoryException e)
{
// never happens because the filter is always valid.
}
return null;
}
/**
* Solve a conflict detected when replaying a modify operation.
*
* @param op The operation that triggered the conflict detection.
* @param msg The operation that triggered the conflict detection.
* @return true if the process is completed, false if it must continue..
*/
{
{
/*
* The operation is a modification but
* the entry has been renamed on a different master in the same time.
* search if the entry has been renamed, and return the new dn
* of the entry.
*/
{
// There is an entry with the same unique id as this modify operation
// replay the modify using the current dn of this entry.
return false;
}
else
{
// This entry does not exist anymore.
// It has probably been deleted, stop the processing of this operation
return true;
}
}
else
{
// The other type of errors can not be caused by naming conflicts.
// Log a message for the repair tool.
return true;
}
}
/**
* Solve a conflict detected when replaying a delete operation.
*
* @param op The operation that triggered the conflict detection.
* @param msg The operation that triggered the conflict detection.
* @return true if the process is completed, false if it must continue..
*/
{
{
/*
* Find if the entry is still in the database.
*/
{
/*
* The entry has already been deleted, either because this delete
* has already been replayed or because another concurrent delete
* has already done the job.
* In any case, there is is nothing more to do.
*/
return true;
}
else
{
/*
* This entry has been renamed, replay the delete using its new DN.
*/
return false;
}
}
{
/*
* This may happen when we replay a DELETE done on a master
* but children of this entry have been added on another master.
*
* Rename all the children by adding entryuuid in dn and delete this entry.
*
* The action taken here must be consistent with the actions
* done in the solveNamingConflict(AddOperation) method
* when we are adding an entry whose parent entry has already been deleted.
*/
return false;
}
else
{
// The other type of errors can not be caused by naming conflicts.
// Log a message for the repair tool.
return true;
}
}
/**
* Solve a conflict detected when replaying a Modify DN operation.
*
* @param op The operation that triggered the conflict detection.
* @param msg The operation that triggered the conflict detection.
* @return true if the process is completed, false if it must continue.
* @throws Exception When the operation is not valid.
*/
{
/*
* four possible cases :
* - the modified entry has been renamed
* - the new parent has been renamed
* - the operation is replayed for the second time.
* - the entry has been deleted
* action :
* - change the target dn and the new parent dn and
* restart the operation,
* - don't do anything if the operation is replayed.
*/
// Construct the new DN to use for the entry.
if (newSuperior == null)
{
}
else
{
}
{
/* this should never happen
* can't solve any conflict in this case.
*/
throw new Exception("operation parameters are invalid");
}
// get the current DN of this entry in the database.
{
// The entry targetted by the Modify DN is not in the database
// anymore.
// This is a conflict between a delete and this modify DN.
// The entry has been deleted anymore so we can safely assume
// that the operation is completed.
return true;
}
// if the newDN and the current DN match then the operation
// is a no-op (this was probably a second replay)
// don't do anything.
{
return true;
}
// If we could not find the new parent entry, we missed this entry
// earlier or it has disappeared from the database
// Log this information for the repair tool and mark the entry
// as conflicting.
// stop the processing.
if (newSuperior == null)
{
return true;
}
{
/*
* The entry or it's new parent has not been found
* reconstruct the operation with the DN we just built
*/
return false;
}
{
/*
* This may happen when two modifyDn operation
* are done on different servers but with the same target DN
* add the conflict object class to the entry
* and rename it using its entryuuid.
*/
modifyDnMsg.getNewRDN()));
return false;
}
else
{
// The other type of errors can not be caused by naming conflicts.
// Log a message for the repair tool.
return true;
}
}
/**
* Solve a conflict detected when replaying a ADD operation.
*
* @param op The operation that triggered the conflict detection.
* @param msg The message that triggered the conflict detection.
* @return true if the process is completed, false if it must continue.
* @throws Exception When the operation is not valid.
*/
{
{
/*
* This can happen if the parent has been renamed or deleted
* find the parent dn and calculate a new dn for the entry
*/
if (parentUniqueId == null)
{
/*
* This entry is the base dn of the backend.
* It is quite surprising that the operation result be NO_SUCH_OBJECT.
* There is nothing more we can do except TODO log a
* message for the repair tool to look at this problem.
*/
return true;
}
{
/*
* The parent has been deleted
* rename the entry as a conflicting entry.
* The action taken here must be consistent with the actions
* done when in the solveNamingConflict(DeleteOperation) method
* when we are deleting an entry that have some child entries.
*/
+ baseDN);
// reset the parent uid so that the check done is the handleConflict
// phase does not fail.
return false;
}
else
{
return false;
}
}
{
/*
* This can happen if
* - two adds are done on different servers but with the
* same target DN.
* - the same ADD is being replayed for the second time on this server.
* if the nsunique ID already exist, assume this is a replay and
* don't do anything
* if the entry unique id do not exist, generate conflict.
*/
{
// entry already exist : this is a replay
return true;
}
else
{
return false;
}
}
else
{
// The other type of errors can not be caused by naming conflicts.
// log a message for the repair tool.
return true;
}
}
/**
* Find all the entries below the provided DN and rename them
* so that they stay below the baseDn of this replicationDomain and
* use the conflicting name and attribute.
*
* @param entryUid The unique ID of the entry whose child must be renamed.
* @param entryDN The DN of the entry whose child must be renamed.
* @param conflictOp The Operation that generated the conflict.
*/
private void findAndRenameChild(
{
// Find an rename child entries.
try
{
attrs);
{
{
{
}
}
}
else
{
// log error and information for the REPAIR tool.
}
} catch (DirectoryException e)
{
// log errror and information for the REPAIR tool.
}
}
/**
* Rename an entry that was conflicting so that it stays below the
* baseDN of the replicationDomain.
*
* @param conflictOp The Operation that caused the conflict.
* @param dn The DN of the entry to be renamed.
* @param uid The uniqueID of the entry to be renamed.
*/
{
{
// log information for the repair tool.
}
}
/**
* Generate a modification to add the conflict attribute to an entry
* whose Dn is now conflicting with another entry.
*
* @param op The operation causing the conflict.
* @param currentDN The current DN of the operation to mark as conflicting.
* @param conflictDN The newDn on which the conflict happened.
*/
{
// create new internal modify operation and run it.
{
// Log information for the repair tool.
}
// Generate an alert to let the administratot know that some
// conflict could not be solved.
}
/**
* Add the conflict attribute to an entry that could
* not be added because it is conflicting with another entry.
*
* @param msg The conflicting Add Operation.
*
* @throws ASN1Exception When an encoding error happenned manipulating the
* msg.
*/
{
// Generate an alert to let the administratot know that some
// conflict could not be solved.
// Add the conflict attribute
}
/**
* Generate the Dn to use for a conflicting entry.
*
* @param entryUid The unique identifier of the entry involved in the
* conflict.
* @param rdn Original rdn.
* @return The generated RDN for a conflicting entry.
*/
{
}
/**
* Generate the RDN to use for a conflicting entry whose father was deleted.
*
* @param entryUid The unique identifier of the entry involved in the
* conflict.
* @param dn The original DN of the entry.
*
* @return The generated RDN for a conflicting entry.
* @throws DirectoryException
*/
{
try
{
} catch (DirectoryException e)
{
// cannot happen
}
return rdn;
}
/**
* Check if an operation must be processed as an assured operation.
*
* @param op the operation to be checked.
* @return true if the operations must be processed as an assured operation.
*/
{
// TODO : should have a filtering mechanism for checking
// operation that are assured and operations that are not.
return false;
}
/**
* Get the maximum receive window size.
*
* @return The maximum receive window size.
*/
public int getMaxRcvWindow()
{
return broker.getMaxRcvWindow();
else
return 0;
}
/**
* Get the current receive window size.
*
* @return The current receive window size.
*/
public int getCurrentRcvWindow()
{
return broker.getCurrentRcvWindow();
else
return 0;
}
/**
* Get the maximum send window size.
*
* @return The maximum send window size.
*/
public int getMaxSendWindow()
{
return broker.getMaxSendWindow();
else
return 0;
}
/**
* Get the current send window size.
*
* @return The current send window size.
*/
public int getCurrentSendWindow()
{
return broker.getCurrentSendWindow();
else
return 0;
}
/**
* Get the number of times the replication connection was lost.
* @return The number of times the replication connection was lost.
*/
public int getNumLostConnections()
{
return broker.getNumLostConnections();
else
return 0;
}
/**
* Get the number of modify conflicts successfully resolved.
* @return The number of modify conflicts successfully resolved.
*/
public int getNumResolvedModifyConflicts()
{
return numResolvedModifyConflicts.get();
}
/**
* Get the number of namign conflicts successfully resolved.
* @return The number of naming conflicts successfully resolved.
*/
public int getNumResolvedNamingConflicts()
{
return numResolvedNamingConflicts.get();
}
/**
* Get the number of unresolved conflicts.
* @return The number of unresolved conflicts.
*/
public int getNumUnresolvedNamingConflicts()
{
return numUnresolvedNamingConflicts.get();
}
/**
* Get the server ID.
* @return The server ID.
*/
public int getServerId()
{
return serverId;
}
/**
* Check if the domain solve conflicts.
*
* @return a boolean indicating if the domain should sove conflicts.
*/
public boolean solveConflict()
{
return solveConflictFlag;
}
/**
* Disable the replication on this domain.
* The session to the replication server will be stopped.
* The domain will not be destroyed but call to the pre-operation
* methods will result in failure.
* The listener thread will be destroyed.
* The monitor informations will still be accessible.
*/
public void disable()
{
disabled = true;
// Stop the listener thread
// Wait for the listener thread to stop
}
/**
* Enable back the domain after a previous disable.
* The domain will connect back to a replication Server and
* will recreate threads to listen for messages from the Sycnhronization
* server.
* The generationId will be retrieved or computed if necessary.
* The ServerState will also be read again from the local database.
*/
public void enable()
{
disabled = false;
try
{
}
catch (Exception e)
{
/* TODO should mark that replicationServer service is
* not available, log an error and retry upon timeout
* should we stop the modifications ?
*/
return;
}
// After an on-line import, the value of the generationId is new
// and it is necessary for the broker to send this new value as part
// of the serverStart message.
// Create the listener thread
}
/**
* Compute the data generationId associated with the current data present
* in the backend for this domain.
* @return The computed generationId.
* @throws DirectoryException When an error occurs.
*/
public long computeGenerationId() throws DirectoryException
{
this.acquireIEContext();
ieContext.checksumOutput = true;
if (debugEnabled())
ieContext.checksumOutput = false;
this.releaseIEContext();
return genId;
}
/**
* Returns the generationId set for this domain.
*
* @return The generationId.
*/
public long getGenerationId()
{
return generationId;
}
/**
* The attribute name used to store the state in the backend.
*/
protected static final String REPLICATION_GENERATION_ID =
"ds-sync-generation-id";
/**
* Stores the value of the generationId.
* @param generationId The value of the generationId.
* @return a ResultCode indicating if the method was successfull.
*/
{
// The generationId is stored in the root entry of the domain.
mods);
op.setInternalOperation(true);
op.setSynchronizationOperation(true);
op.setDontSynchronize(true);
{
generationIdSavedStatus = false;
{
// The case where the backend is empty (NO_SUCH_OBJECT)
// is not an error case.
}
}
else
{
generationIdSavedStatus = true;
}
return result;
}
/**
* Load the GenerationId from the root entry of the domain
* from the REPLICATION_GENERATION_ID attribute in database
* to memory, or compute it if not found.
*
* @return generationId The retrieved value of generationId
* @throws DirectoryException When an error occurs.
*/
public long loadGenerationId()
throws DirectoryException
{
long generationId=-1;
if (debugEnabled())
boolean found = false;
try
{
}
catch (LDAPException e)
{
// can not happen
return -1;
}
/*
* Search the database entry that is used to periodically
* save the ServerState
*/
{
}
{
if (resultEntry != null)
{
{
{
" Must be exactly 1 in entry " +
}
{
found=true;
try
{
getStringValue());
}
catch(Exception e)
{
}
}
}
}
}
if (!found)
{
if (debugEnabled())
" generationId=" + generationId);
}
else
{
generationIdSavedStatus = true;
if (debugEnabled())
"Generation ID successfully read from domain base DN=" + baseDN +
" generationId=" + generationId);
}
return generationId;
}
/**
* 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.
*
* @param generationIdNewValue The new value of the generation Id.
*/
{
if (debugEnabled())
if (generationIdNewValue == null)
{
}
else
{
}
}
/**
* Do whatever is needed when a backup is started.
* We need to make sure that the serverState is correclty save.
*/
public void backupStart()
{
}
/**
* Do whatever is needed when a backup is finished.
*/
public void backupEnd()
{
// Nothing is needed at the moment
}
/*
* Total Update >>
*/
/**
* Receives bytes related to an entry in the context of an import to
* initialize the domain (called by ReplLDIFInputStream).
*
* @return The bytes. Null when the Done or Err message has been received
*/
public byte[] receiveEntryBytes()
{
while (true)
{
try
{
if (debugEnabled())
" sid:" + this.serverId +
" base DN:" + this.baseDN +
" Import EntryBytes received " + msg);
{
// The server is in the shutdown process
return null;
}
if (msg instanceof EntryMessage)
{
return entryBytes;
}
else if (msg instanceof DoneMessage)
{
// This is the normal termination of the import
// No error is stored and the import is ended
// by returning null
return null;
}
else if (msg instanceof ErrorMessage)
{
// This is an error termination during the import
// The error is stored and the import is ended
// by returning null
errorMsg.getDetails());
return null;
}
else
{
// Other messages received during an import are trashed
}
}
catch(Exception e)
{
// TODO: i18n
e.getLocalizedMessage()));
}
}
}
/**
* on going.
* @param errorMsg The error message received.
*/
{
// FIXME TBD Treat the case where the error happens while entries
// are being exported
if (debugEnabled())
" abandonImportExport:" + this.serverId +
" base DN:" + this.baseDN +
" Error Msg received " + errorMsg);
{
errorMsg.getDetails());
{
// Update the task that initiated the import
}
}
}
/**
* Clears all the entries from the JE backend determined by the
* be id passed into the method.
*
* @param createBaseEntry Indicate whether to automatically create the base
* entry and add it to the backend.
* @param beID The be id to clear.
* @param dn The suffix of the backend to create if the the createBaseEntry
* boolean is true.
* @throws Exception If an unexpected problem occurs.
*/
{
// FIXME Should setBackendEnabled be part of TaskUtils ?
try
{
{
}
try
{
}
finally
{
}
}
finally
{
}
if (createBaseEntry)
{
}
}
/**
* Export the entries from the backend.
* The ieContext must have been set before calling.
*
* @throws DirectoryException when an error occurred
*/
protected void exportBackend()
throws DirectoryException
{
// Acquire a shared lock for the backend.
try
{
{
throw new DirectoryException(
}
}
catch (Exception e)
{
throw new DirectoryException(
}
if (ieContext.checksumOutput)
{
try
{
getBytes());
}
catch(Exception e)
{
// Should never happen
}
}
else
{
}
// baseDN branch is the only one included in the export
// For the checksum computing mode, only consider the 'stable' attributes
if (ieContext.checksumOutput)
{
{"objectclass", "sn", "cn", "entryuuid"};
{
{
}
}
}
// Launch the export.
try
{
}
catch (DirectoryException de)
{
{
// This is the normal end when computing the generationId
// We can interrupt the export only by an IOException
}
else
{
throw new DirectoryException(
}
}
catch (Exception e)
{
throw new DirectoryException(
}
finally
{
{
}
else
{
// Clean up after the export by closing the export config.
// Will also flush the export and export the remaining entries.
// This is a real export where writer has been initialized.
}
// Release the shared lock on the backend.
try
{
{
throw new DirectoryException(
}
}
catch (Exception e)
{
throw new DirectoryException(
}
}
}
/**
* Retrieves the backend related to the domain.
*
* @return The backend of that domain.
* @param baseDN The baseDN to retrieve the backend
*/
{
// Retrieves the backend related to this domain
}
/**
* Get the internal broker to perform some operations on it.
*
* @return The broker for this domain.
*/
{
return broker;
}
/**
* Exports an entry in LDIF format.
*
* @param lDIFEntry The entry to be exported..
*
* @throws IOException when an error occurred.
*/
{
// If an error was raised - like receiving an ErrorMessage
// we just let down the export.
{
throw ioe;
}
if (ieContext.checksumOutput == false)
{
}
try
{
}
catch (DirectoryException de)
{
}
}
/**
* Initializes this domain from another source server.
*
* @param source The source from which to initialize
* @param initTask The task that launched the initialization
* and should be updated of its progress.
* @throws DirectoryException when an error occurs
*/
throws DirectoryException
{
if (debugEnabled())
// Publish Init request msg
// .. we expect to receive entries or err after that
}
/**
* Verifies that the given string represents a valid source
* from which this server can be initialized.
* @param sourceString The string representing the source
* @return The source as a short value
* @throws DirectoryException if the string is not valid
*/
throws DirectoryException
{
short source = 0;
try
{
{
// TODO Verifies serverID is in the domain
// We shold check here that this is a server implied
// in the current domain.
return source;
}
}
catch(Exception e)
{
cause = e;
}
{
throw new DirectoryException(
}
else
{
throw new DirectoryException(
}
}
/**
* Verifies that the given string represents a valid source
* from which this server can be initialized.
* @param targetString The string representing the source
* @return The source as a short value
* @throws DirectoryException if the string is not valid
*/
throws DirectoryException
{
short target = 0;
{
return RoutableMessage.ALL_SERVERS;
}
// So should be a serverID
try
{
if (target >= 0)
{
// FIXME Could we check now that it is a know server in the domain ?
}
return target;
}
catch(Exception e)
{
cause = e;
}
throw new DirectoryException(
else
throw new DirectoryException(
}
private synchronized void acquireIEContext()
throws DirectoryException
{
{
// Rejects 2 simultaneous exports
message);
}
}
private synchronized void releaseIEContext()
{
}
/**
* Process the initialization of some other server or servers in the topology
* specified by the target argument.
* @param target The target that should be initialized
* @param initTask The task that triggers this initialization and that should
* be updated with its progress.
*
* @exception DirectoryException When an error occurs.
*/
throws DirectoryException
{
}
/**
* 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.
*
* @param target The target that should be initialized.
* @param requestorID The server that initiated the export.
* @param initTask The task that triggers this initialization and that should
* be updated with its progress.
*
* @exception DirectoryException When an error occurs.
*/
throws DirectoryException
{
try
{
if (!backend.supportsLDIFExport())
{
}
// The number of entries to be exported is the number of entries under
// the base DN entry and the base entry itself.
{
}
// Send start message to the peer
// Notify the peer of the success
}
catch(DirectoryException de)
{
// Notify the peer of the failure
new ErrorMessage(target,
de.getMessageObject());
throw(de);
}
}
/**
* Process backend before import.
* @param backend The backend.
* @throws Exception
*/
throws Exception
{
// Stop saving state
stateSavingDisabled = true;
// FIXME setBackendEnabled should be part of TaskUtils ?
// Acquire an exclusive lock for the backend.
{
}
}
/**
* Initializes the domain's backend with received entries.
* @param initializeMessage The message that initiated the import.
* @exception DirectoryException Thrown when an error occurs.
*/
throws DirectoryException
{
try
{
if (!backend.supportsLDIFImport())
{
}
else
{
{
// The import responds to a request we did so the IEContext
// is already acquired
}
else
{
}
importConfig.setAppendToExistingData(false);
// TODO How to deal with rejected entries during the import
"replInitRejectedEntries").getAbsolutePath(),
// Process import
stateSavingDisabled = false;
}
}
catch(Exception e)
{
}
finally
{
// Cleanup
if (importConfig != null)
{
// Re-enable backend
}
// Update the task that initiated the import
{
}
}
// Sends up the root error.
{
throw de;
}
else
{
// Retrieves the generation ID associated with the data imported
try
{
}
catch (DirectoryException e)
{
e.getLocalizedMessage()));
}
if (debugEnabled())
"After import, the replication plugin restarts connections" +
" to all RSs to provide new generation ID=" + generationId);
// Re-exchange generationID and state with RS
}
}
/**
* Make post import operations.
* @param backend The backend implied in the import.
* @exception DirectoryException Thrown when an error occurs.
*/
throws DirectoryException
{
// Release lock
{
}
}
/**
* Retrieves a replication domain based on the baseDN.
*
* @param baseDN The baseDN of the domain to retrieve
* @return The domain retrieved
* @throws DirectoryException When an error occurred or no domain
* match the provided baseDN.
*/
throws DirectoryException
{
// Retrieves the domain
for (SynchronizationProvider provider :
{
if (!( provider instanceof MultimasterReplication))
{
message);
}
// From the domainDN retrieves the replication domain
{
break;
}
if (replicationDomain != null)
{
// Should never happen
message);
}
}
if (replicationDomain == null)
{
}
return replicationDomain;
}
/**
* Returns the backend associated to this domain.
* @return The associated backend.
*/
public Backend getBackend()
{
return retrievesBackend(baseDN);
}
/**
* Returns a boolean indiciating if an import or export is currently
* processed.
* @return The status
*/
public boolean ieRunning()
{
}
/*
* <<Total Update
*/
/**
* Push the modifications contain the in given parameter has
* a modification that would happen on a local server.
* The modifications are not applied to the local database,
* historical information is not updated but a ChangeNumber
* is generated and the ServerState associated to this domain is
* updated.
* @param modifications The modification to push
*/
{
}
/**
* Check if the provided configuration is acceptable for add.
*
* @param configuration The configuration to check.
* @param unacceptableReasons When the configuration is not acceptable, this
* table is use to return the reasons why this
* configuration is not acceptbale.
*
* @return true if the configuration is acceptable, false other wise.
*/
public static boolean isConfigurationAcceptable(
{
// Check that there is not already a domain with the same DN
{
return false;
}
// Check that the base DN is configured as a base-dn of the directory server
{
return false;
}
return true;
}
/**
* {@inheritDoc}
*/
{
// server id and base dn are readonly.
// isolationPolicy can be set immediately and will apply
// to the next updates.
// The other parameters needs to be renegociated with the ReplicationServer.
// so that requires restarting the session with the ReplicationServer.
}
/**
* {@inheritDoc}
*/
public boolean isConfigurationChangeAcceptable(
{
return true;
}
/**
* {@inheritDoc}
*/
{
return alerts;
}
/**
* {@inheritDoc}
*/
public String getClassName()
{
return CLASS_NAME;
}
/**
* {@inheritDoc}
*/
public DN getComponentEntryDN()
{
return configDn;
}
/**
* Check if the domain is connected to a ReplicationServer.
*
* @return true if the server is connected, false if not.
*/
public boolean isConnected()
{
return broker.isConnected();
else
return false;
}
/**
* Determine whether the connection to the replication server is encrypted.
* @return true if the connection is encrypted, false otherwise.
*/
public boolean isSessionEncrypted()
{
return broker.isSessionEncrypted();
else
return false;
}
}