LDAPReplicationDomain.java revision cf364c082dfe5ea566abc3c20bc5546a4629c5eb
/*
* 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 legal-notices/CDDLv1_0.txt
* 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 legal-notices/CDDLv1_0.txt.
* 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-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2015 ForgeRock AS
*/
/**
* This class implements the bulk part 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.
* <p>
* FIXME Move this class to org.opends.server.replication.service
* or the equivalent package once this code is moved to a maven module.
*/
public final class LDAPReplicationDomain extends ReplicationDomain
implements ConfigurationChangeListener<ReplicationDomainCfg>,
{
/**
* Set of attributes that will return all the user attributes and the
* replication related operational attributes when used in a search operation.
*/
/**
* This class is used in the session establishment phase
* when no Replication Server with all the local changes has been found
* and we therefore need to recover them.
* A search is then performed on the database using this
* internalSearchListener.
*/
private class ScanSearchListener implements InternalSearchListener
{
{
}
public void handleInternalSearchEntry(
throws DirectoryException
{
// Build the list of Operations that happened on this entry after startCSN
// and before endCSN and add them to the replayOperations list
{
{
synchronized (replayOperations)
{
}
}
}
}
public void handleInternalSearchReference(
{
// Nothing to do.
}
}
/**
* The fully-qualified name of this class.
*/
.getName();
/**
* 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.
*/
private final DSRSShutdownSync dsrsShutdownSync;
/**
* The update to replay message queue where the listener thread is going to
* push incoming update messages.
*/
/** The number of naming conflicts successfully resolved. */
/** The number of modify conflicts successfully resolved. */
/** The number of unresolved naming conflicts. */
private final AtomicInteger numUnresolvedNamingConflicts =
new AtomicInteger();
/** The number of updates replayed successfully by the replication. */
private final PersistentServerState state;
private volatile boolean generationIdSavedStatus;
/**
* This object is used to store the list of update currently being
* done on the local database.
* Is is useful 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 final PendingChanges pendingChanges;
/**
* It contain the updates that were done on other servers, transmitted by the
* replication server and that are currently replayed.
* <p>
* It is useful to make sure that dependencies between operations are
* correctly fulfilled and to make sure that the ServerState is not updated
* too early.
*/
private final RemotePendingChanges remotePendingChanges;
private boolean solveConflictFlag = true;
private volatile boolean disabled;
private volatile boolean stateSavingDisabled;
/**
* This list is used to temporary store operations that needs to be replayed
* at session establishment time.
*/
private ExternalChangelogDomain eclDomain;
/**
* A boolean indicating if the thread used to save the persistentServerState
* is terminated.
*/
private volatile boolean done = true;
private final ServerStateFlush flushThread;
/**
* The attribute name used to store the generation id in the backend.
*/
private static final String REPLICATION_GENERATION_ID =
"ds-sync-generation-id";
/**
* The attribute name used to store the fractional include configuration in
* the backend.
*/
static final String REPLICATION_FRACTIONAL_INCLUDE =
"ds-sync-fractional-include";
/**
* The attribute name used to store the fractional exclude configuration in
* the backend.
*/
static final String REPLICATION_FRACTIONAL_EXCLUDE =
"ds-sync-fractional-exclude";
/**
* Fractional replication variables.
*/
/** Holds the fractional configuration for this domain, if any. */
private final FractionalConfig fractionalConfig;
/**
* The list of attributes that cannot be used in fractional replication
* configuration.
*/
{
"objectClass",
"2.5.4.0" // objectClass OID
};
/**
* When true, this flag is used to force the domain status to be put in bad
* data set just after the connection to the replication server.
* This must be used when fractional replication is enabled with a
* configuration different from the previous one (or at the very first
* fractional usage time) : after connection, a ChangeStatusMsg is sent
* requesting the bad data set status. Then none of the update messages
* received from the replication server are taken into account until the
* backend is synchronized with brand new data set compliant with the new
* fractional configuration (i.e with compliant fractional configuration in
* domain root entry).
*/
private boolean forceBadDataSet;
/**
* The message id to be used when an import is stopped with error by
* the fractional replication ldif import plugin.
*/
private int importErrorMessageId = -1;
/**
* LocalizableMessage type for ERR_FULL_UPDATE_IMPORT_FRACTIONAL_BAD_REMOTE.
*/
static final int IMPORT_ERROR_MESSAGE_BAD_REMOTE = 1;
/**
* LocalizableMessage type for ERR_FULL_UPDATE_IMPORT_FRACTIONAL_REMOTE_IS_FRACTIONAL.
*/
static final int IMPORT_ERROR_MESSAGE_REMOTE_IS_FRACTIONAL = 2;
/*
* Definitions for the return codes of the
* fractionalFilterOperation(PreOperationModifyOperation
* modifyOperation, boolean performFiltering) method
*/
/**
* The operation contains attributes subject to fractional filtering according
* to the fractional configuration.
*/
private static final int FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES = 1;
/**
* The operation contains no attributes subject to fractional filtering
* according to the fractional configuration.
*/
private static final int FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES = 2;
/** The operation should become a no-op. */
private static final int FRACTIONAL_BECOME_NO_OP = 3;
/**
* The last CSN purged in this domain. Allows to have a continuous purging
* process from one purge processing (task run) to the next one. Values 0 when
* the server starts.
*/
/**
* The thread that periodically saves the ServerState of this
* LDAPReplicationDomain in the database.
*/
private class ServerStateFlush extends DirectoryThread
{
protected ServerStateFlush()
{
}
/** {@inheritDoc} */
public void run()
{
done = false;
while (!isShutdownInitiated())
{
try
{
synchronized (this)
{
wait(1000);
if (!disabled && !stateSavingDisabled)
{
// save the ServerState
}
}
}
catch (InterruptedException e)
{
// Thread interrupted: check for shutdown.
}
}
done = true;
}
}
/**
* The thread that is responsible to update the RS to which this domain is
* connected in case it is late and there is no RS which is up to date.
*/
private class RSUpdater extends DirectoryThread
{
{
super("Replica DS(" + getServerId() + ") missing change publisher for domain \"" + getBaseDN() + "\"");
this.startCSN = replServerMaxCSN;
}
/** {@inheritDoc} */
public void run()
{
// Replication server is missing some of our changes:
// let's send them to him.
/*
* Get all the changes that have not been seen by this
* replication server and publish them.
*/
try
{
{
synchronized(replayOperations)
{
}
}
else
{
/*
* An error happened trying to search for the updates
* This server will start accepting again new updates but
* some inconsistencies will stay between servers.
* Log an error for the repair tool
* that will need to re-synchronize the servers.
*/
}
}
catch (Exception e)
{
/*
* An error happened trying to search for the updates
* This server will start accepting again new updates but
* some inconsistencies will stay between servers.
* Log an error for the repair tool
* that will need to re-synchronize the servers.
*/
}
finally
{
broker.setRecoveryRequired(false);
// RSUpdater thread has finished its work, let's remove it from memory
// so another RSUpdater thread can be started if needed.
}
}
}
/**
* 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.
*/
{
super(configuration, -1);
this.dsrsShutdownSync = dsrsShutdownSync;
// Get assured configuration
readAssuredConfig(configuration, false);
// Get fractional configuration
readFractionalConfig(configuration, false);
{
}
try
{
}
catch (DirectoryException e)
{
}
/*
* Create a new Persistent Server State that will be used to store
* the last CSN seen from all LDAP servers in the topology.
*/
getServerState());
flushThread = new ServerStateFlush();
/*
* CSNGenerator is used to create new unique CSNs for each operation done on
* this replication domain.
*
* The generator time is adjusted to the time of the last CSN received from
* remote other servers.
*/
// listen for changes on the configuration
configuration.addChangeListener(this);
// register as an AlertGenerator
}
/**
* 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.
*/
{
&& cfg.isSolveConflicts();
}
/**
* Sets the error message id to be used when online import is stopped with
* error by the fractional replication ldif import plugin.
* @param importErrorMessageId The message to use.
*/
void setImportErrorMessageId(int importErrorMessageId)
{
}
/**
* This flag is used by the fractional replication ldif import plugin to stop
* the (online) import process if a fractional configuration inconsistency is
* detected by it.
*
* @return true if the online import currently in progress should continue,
* false otherwise.
*/
private boolean isFollowImport()
{
return importErrorMessageId == -1;
}
/**
* Gets and stores the fractional replication configuration parameters.
* @param configuration The configuration object
* @param allowReconnection Tells if one must reconnect if significant changes
* occurred
*/
boolean allowReconnection)
{
// Read the configuration entry
try
{
}
catch(ConfigException e)
{
// Should not happen as normally already called without problem in
// isConfigurationChangeAcceptable or isConfigurationAcceptable
// if we come up to this method
return;
}
/**
* Is there any change in fractional configuration ?
*/
// Compute current configuration
boolean needReconnection;
try
{
}
catch (ConfigException e)
{
// Should not happen
return;
}
// Disable service if configuration changed
if (needRestart)
{
}
// Set new configuration
if (fractionalConfig.isFractional())
{
// Set new fractional configuration values
} else
{
// Reset default values
}
// Reconnect if required
if (needRestart)
{
}
}
/**
* Return true if the fractional configuration stored in the domain root
* entry of the backend is equivalent to the fractional configuration stored
* in the local variables.
*/
private boolean isBackendFractionalConfigConsistent()
{
// Read config stored in domain root entry
if (logger.isTraceEnabled())
{
logger.trace("Attempt to read the potential fractional config in domain root entry " + getBaseDN());
}
// Search the domain root entry that is used to save the generation id
.addAttribute(REPLICATION_GENERATION_ID, REPLICATION_FRACTIONAL_EXCLUDE, REPLICATION_FRACTIONAL_INCLUDE);
{
return false;
}
if (resultEntry == null)
{
/*
* The backend is probably empty: if there is some fractional
* configuration in memory, we do not let the domain being connected,
* otherwise, it's ok
*/
return !fractionalConfig.isFractional();
}
// Now extract fractional configuration if any
// Compare backend and local fractional configuration
}
{
if (resultEntry != null)
{
{
{
return resultEntry;
}
{
String errorMsg = "#Values=" + attr.size() + " Must be exactly 1 in entry " + resultEntry.toLDIFString();
}
}
}
return null;
}
{
{
{
}
}
return null;
}
/**
* Return true if the fractional configuration passed as fractional
* configuration attribute values is equivalent to the fractional
* configuration stored in the local variables.
* @param fractionalConfig The local fractional configuration
* @param exclIt Fractional exclude mode configuration attribute values to
* analyze.
* @param inclIt Fractional include mode configuration attribute values to
* analyze.
* @return True if the fractional configuration passed as fractional
* configuration attribute values is equivalent to the fractional
* configuration stored in the local variables.
*/
static boolean isFractionalConfigConsistent(
{
/*
* Parse fractional configuration stored in passed fractional configuration
* attributes values
*/
int storedFractionalMode;
try
{
} catch (ConfigException e)
{
// Should not happen as configuration in domain root entry is flushed
// from valid configuration in local variables
return false;
}
// Set stored fractional configuration values
{
}
/*
* Compare configuration stored in passed fractional configuration
* attributes with local variable one
*/
try
{
return FractionalConfig.
} catch (ConfigException e)
{
// Should not happen as configuration in domain root entry is flushed
// from valid configuration in local variables so both should have already
// been checked
return false;
}
}
/**
* Utility class to have get a string iterator from an AtributeValue iterator.
* Assuming the attribute values are strings.
*/
{
/**
* Creates a new AttributeValueStringIterator object.
* @param attrValIt The underlying attribute iterator to use, assuming
* internal values are strings.
*/
{
}
/** {@inheritDoc} */
public boolean hasNext()
{
}
/** {@inheritDoc} */
{
}
/** {@inheritDoc} */
// Should not be needed anyway
public void remove()
{
}
}
/**
* Compare 2 attribute collections and returns true if they are equivalent.
*
* @param attributes1
* First attribute collection to compare.
* @param attributes2
* Second attribute collection to compare.
* @return True if both attribute collection are equivalent.
* @throws ConfigException
* If some attributes could not be retrieved from the schema.
*/
private static boolean areAttributesEquivalent(
throws ConfigException
{
// Compare all classes attributes
{
return false;
}
// Check consistency of all classes attributes
/*
* For each attribute in attributes1, check there is the matching
* one in attributes2.
*/
{
// Get attribute from attributes1
if (attributeType1 == null)
{
throw new ConfigException(
}
// Look for matching one in attributes2
boolean foundAttribute = false;
{
if (attributeType2 == null)
{
throw new ConfigException(
}
{
foundAttribute = true;
break;
}
}
// Found matching attribute ?
if (!foundAttribute)
{
return false;
}
}
return true;
}
/**
* Check that the passed fractional configuration is acceptable
* regarding configuration syntax, schema constraints...
* Throws an exception if the configuration is not acceptable.
* @param configuration The configuration to analyze.
* @throws org.opends.server.config.ConfigException if the configuration is
* not acceptable.
*/
private static void isFractionalConfigAcceptable(
{
/*
* Parse fractional configuration
*/
// Read the configuration entry
if (!newFractionalConfig.isFractional())
{
// Nothing to check
return;
}
// Prepare variables to be filled with config
/*
* Check attributes consistency : we only allow to filter MAY (optional)
* attributes of a class : to be compliant with the schema, no MUST
* (mandatory) attribute can be filtered by fractional replication.
*/
// Check consistency of specific classes attributes
{
// Does the class exist ?
className.toLowerCase());
if (fractionalClass == null)
{
throw new ConfigException(
}
boolean isExtensibleObjectClass =
{
// Not a prohibited attribute ?
{
throw new ConfigException(
}
// Does the attribute exist ?
if (attributeType != null)
{
// No more checking for the extensibleObject class
// Exclusive mode : the attribute must be optional
{
throw new ConfigException(
className));
}
}
else
{
throw new ConfigException(
}
}
}
// Check consistency of all classes attributes
{
// Not a prohibited attribute ?
{
throw new ConfigException(
}
// Does the attribute exist ?
{
throw new ConfigException(
}
}
}
/**
* Test if the passed attribute is not allowed to be used in configuration of
* fractional replication.
* @param attr Attribute to test.
* @return true if the attribute is prohibited.
*/
{
{
{
return true;
}
}
return false;
}
/**
* If fractional replication is enabled, this analyzes the operation and
* suppresses the forbidden attributes in it so that they are not added in
* the local backend.
*
* @param addOperation The operation to modify based on fractional
* replication configuration
* @param performFiltering Tells if the effective attribute filtering should
* be performed or if the call is just to analyze if there are some
* attributes filtered by fractional configuration
* @return true if the operation contains some attributes subject to filtering
* by the fractional configuration
*/
private boolean fractionalFilterOperation(
{
}
/**
* If fractional replication is enabled, this analyzes the operation and
* suppresses the forbidden attributes in it so that they are not added in
* the local backend.
*
* @param modifyDNOperation The operation to modify based on fractional
* replication configuration
* @param performFiltering Tells if the effective modifications should
* be performed or if the call is just to analyze if there are some
* inconsistency with fractional configuration
* @return true if the operation is inconsistent with fractional
* configuration
*/
private boolean fractionalFilterOperation(
{
// Quick exit if not called for analyze and
{
// The core will remove any occurrence of attribute that was part of the
// old RDN, nothing more to do.
return true; // Will not be used as analyze was not requested
}
// Create a list of filtered attributes for this entry
{
// No attributes to filter
return false;
}
/*
* Analyze the old and new rdn to see if they are some attributes to be
* removed: if the oldRDN contains some forbidden attributes (for instance
* it is possible if the entry was created with an add operation and the
* RDN used contains a forbidden attribute: in this case the attribute value
* has been kept to be consistent with the dn of the entry.) that are no
* more part of the new RDN, we must remove any attribute of this type by
* putting a modification to delete the attribute.
*/
boolean inconsistentOperation = false;
// Go through each attribute of the old RDN
{
// Is it present in the fractional attributes established list ?
boolean foundAttribute =
&& !modifyDNOperation.deleteOldRDN())
{
/*
* A forbidden attribute is in the old RDN and no more in the new RDN,
* and it has not been requested to remove attributes from old RDN:
* let's remove the attribute from the entry to stay consistent with
* fractional configuration
*/
inconsistentOperation = true;
}
}
return inconsistentOperation;
}
{
{
{
return true;
}
}
return false;
}
/**
* Remove attributes from an entry, according to the passed fractional
* configuration. The entry is represented by the 2 passed parameters.
* The attributes to be removed are removed using the remove method on the
* passed iterator for the attributes in the entry.
* @param fractionalConfig The fractional configuration to use
* @param entryRdn The rdn of the entry to add
* @param classes The object classes representing the entry to modify
* @param attributesMap The map of attributes/values to be potentially removed
* from the entry.
* @param performFiltering Tells if the effective attribute filtering should
* be performed or if the call is just an analyze to see if there are some
* attributes filtered by fractional configuration
* @return true if the operation contains some attributes subject to filtering
* by the fractional configuration
*/
static boolean fractionalRemoveAttributesFromEntry(
{
boolean hasSomeAttributesToFilter = false;
/*
* fractional replication configuration
*/
{
return false; // No attributes to filter
}
// Prepare list of object classes of the added entry
/*
* Go through the user attributes and remove those that match filtered one
* - exclude mode : remove only attributes that are in
* fractionalConcernedAttributes
* - include mode : remove any attribute that is not in
* fractionalConcernedAttributes
*/
{
// Only optional attributes may be removed
// Do not remove an attribute if it is a prohibited one
{
continue;
}
if (!performFiltering)
{
// The call was just to check : at least one attribute to filter
// found, return immediately the answer;
return true;
}
// entry as it is forbidden
{
/*
We must remove all values of the attributes map for this
attribute type but the one that has the value which is in the RDN
of the entry. In fact the (underlying )attribute list does not
support remove so we have to create a new list, keeping only the
attribute value which is the same as in the RDN
*/
// Locate the attribute value identical to the one in the RDN
{
{
// Keep the value we want
} else {
hasSomeAttributesToFilter = true;
}
}
}
else
{
hasSomeAttributesToFilter = true;
}
}
// Recreate the attribute list with only the RDN attribute value
if (sameAttrValue != null)
// Paranoia check: should never be the case as we should always
{
// Construct and store new attribute list
/*
Store matching attribute type
The mapping will be done using object from rdnAttrTypes as key
and object from newRdnAttrLists (at same index) as value in
the user attribute map to be modified
*/
}
}
else
{
// Found an attribute to remove, remove it from the list.
hasSomeAttributesToFilter = true;
}
}
// Now overwrite the attribute values for the attribute types present in the
// RDN, if there are some filtered attributes in the RDN
{
}
return hasSomeAttributesToFilter;
}
{
return list;
}
{
return list;
}
{
{
{
return true;
}
}
return false;
}
{
}
{
// Is the current attribute part of the established list ?
boolean foundAttribute =
// Now remove the attribute or modification if:
// - exclusive mode and attribute is in configuration
// - inclusive mode and attribute is not in configuration
}
private static boolean canRemoveAttribute(boolean fractionalExclusive,
boolean foundAttribute)
{
return (foundAttribute && fractionalExclusive)
|| (!foundAttribute && !fractionalExclusive);
}
{
}
/**
* Prepares a list of attributes of interest for the fractional feature.
* @param fractionalConfig The fractional configuration to use
* @param entryObjectClasses The object classes of an entry on which an
* operation is going to be performed.
* when the operation will be performed.
*/
{
/*
* Is the concerned entry of a type concerned by fractional replication
* configuration ? If yes, add the matching attribute names to a set of
* attributes to take into account for filtering
* (inclusive or exclusive mode).
* Using a Set to avoid duplicate attributes (from 2 inheriting classes for
* instance)
*/
// Get object classes the entry matches
{
{
{
}
}
}
// Add to the set any attribute which is class independent
return fractionalConcernedAttributes;
}
/**
* If fractional replication is enabled, this analyzes the operation and
* suppresses the forbidden attributes in it so that they are not added/
*
* @param modifyOperation The operation to modify based on fractional
* replication configuration
* @param performFiltering Tells if the effective attribute filtering should
* be performed or if the call is just to analyze if there are some
* attributes filtered by fractional configuration
* @return FRACTIONAL_HAS_FRACTIONAL_FILTERED_ATTRIBUTES,
* FRACTIONAL_HAS_NO_FRACTIONAL_FILTERED_ATTRIBUTES or FRACTIONAL_BECOME_NO_OP
*/
private int fractionalFilterOperation(PreOperationModifyOperation
modifyOperation, boolean performFiltering)
{
/*
* fractional replication configuration
*/
{
// No attributes to filter
}
// Prepare list of object classes of the modified entry
try
{
}
catch(DirectoryException e)
{
}
/*
* Now go through the attribute modifications and filter the mods according
* to the fractional configuration (using the just established concerned
* attributes list):
* - delete attributes: remove them if regarding a filtered attribute
* - add attributes: remove them if regarding a filtered attribute
* - modify attributes: remove them if regarding a filtered attribute
*/
{
// Fractional replication ignores operational attributes
if (attrType.isOperational()
{
continue;
}
if (!performFiltering)
{
// The call was just to check : at least one attribute to filter
// found, return immediately the answer;
}
// Found a modification to remove, remove it from the list.
{
// This operation must become a no-op as no more modification in it
return FRACTIONAL_BECOME_NO_OP;
}
}
return result;
}
/**
* This is overwritten to allow stopping the (online) import process by the
* fractional ldif import plugin when it detects that the (imported) remote
* data set is not consistent with the local fractional configuration.
* {@inheritDoc}
*/
protected byte[] receiveEntryBytes()
{
if (isFollowImport())
{
// Ok, next entry is allowed to be received
return super.receiveEntryBytes();
}
// Fractional ldif import plugin detected inconsistency between local and
// remote server fractional configuration and is stopping the import
// process:
// This is an error termination during the import
// The error is stored and the import is ended by returning null
switch (importErrorMessageId)
{
break;
msg = NOTE_ERR_FULL_UPDATE_IMPORT_FRACTIONAL_REMOTE_IS_FRACTIONAL.get(getBaseDN(), ieCtx.getImportSource());
break;
}
return null;
}
/**
* This is overwritten to allow stopping the (online) export process if the
* local domain is fractional and the destination is all other servers:
* This make no sense to have only fractional servers in a replicated
* topology. This prevents from administrator manipulation error that would
* lead to whole topology data corruption.
* {@inheritDoc}
*/
{
{
LocalizableMessage msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_FULL_UPDATE_FRACTIONAL.get(getBaseDN(), getServerId());
}
// FIXME should the next call use the initWindow parameter rather than the
// instance variable?
}
/**
* Implement the handleConflictResolution phase of the deleteOperation.
*
* @param deleteOperation The deleteOperation.
* @return A SynchronizationProviderResult indicating if the operation
* can continue.
*/
{
{
return new SynchronizationProviderResult.StopProcessing(
}
{
/*
* This is a replication operation
* Check that the modified entry has the same entryuuid
* as it 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_OBJECT.
* 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.StopProcessing(
}
}
else
{
// There is no replication context attached to the operation
// so this is not a replication operation.
synchronized (replayOperations)
{
if (size >= 10000)
{
}
}
}
return new SynchronizationProviderResult.ContinueProcessing();
}
/**
* Implement the handleConflictResolution phase of the addOperation.
*
* @param addOperation The AddOperation.
* @return A SynchronizationProviderResult indicating if the operation
* can continue.
*/
{
{
return new SynchronizationProviderResult.StopProcessing(
}
if (fractionalConfig.isFractional())
{
{
/*
* Filter attributes here for fractional replication. If fractional
* replication is enabled, we analyze the operation to suppress the
* forbidden attributes in it so that they are not added in the local
* backend. This must be called before any other plugin is called, to
* keep coherency across plugin calls.
*/
}
else
{
/*
* Direct access from an LDAP client : if some attributes are to be
* removed according to the fractional configuration, simply forbid
* the operation
*/
if (fractionalFilterOperation(addOperation, false))
{
return new SynchronizationProviderResult.StopProcessing(
}
}
}
{
/*
* If an entry with the same entry uniqueID already exist then
* this operation has already been replayed in the past.
*/
{
return new SynchronizationProviderResult.StopProcessing(
}
/* 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.
if (parentEntryUUID != null)
{
// 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.StopProcessing(
}
if (parentDnFromEntryDn != null
{
// parentEntry has been renamed
// replication name conflict resolution is expected to fix that
// later in the flow
return new SynchronizationProviderResult.StopProcessing(
}
}
}
return new SynchronizationProviderResult.ContinueProcessing();
}
/**
* 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, the response message,
* interrupt the operation, and return false
*
* @return true when it OK to process the Operation, false otherwise.
* When false is returned the resultCode and the response message
* is also set in the Operation.
*/
private boolean brokerIsConnected()
{
{
// this policy imply that we always accept updates.
return true;
}
{
// this isolation policy specifies that the updates are denied
// when the broker had problems during the connection phase
// Updates are still accepted if the broker is currently connecting..
return !hasConnectionError();
}
// 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.
*/
{
{
return new SynchronizationProviderResult.StopProcessing(
}
if (fractionalConfig.isFractional())
{
{
/*
* Filter operation here for fractional replication. If fractional
* replication is enabled, we analyze the operation and modify it if
* necessary to stay consistent with what is defined in fractional
* configuration.
*/
}
else
{
/*
* Direct access from an LDAP client : something is inconsistent with
* the fractional configuration, forbid the operation.
*/
if (fractionalFilterOperation(modifyDNOperation, false))
{
LocalizableMessage msg = NOTE_ERR_FRACTIONAL_FORBIDDEN_OPERATION.get(getBaseDN(), modifyDNOperation);
return new SynchronizationProviderResult.StopProcessing(
}
}
}
{
/*
* This is a replication operation
* Check that the modified entry has the same entryuuid
* as was in the original message.
*/
final String modifiedEntryUUID =
{
/*
* 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_OBJECT.
* 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.StopProcessing(
}
{
/*
* Also check that the current id of the
* parent is the same as when the operation was performed.
*/
{
return new SynchronizationProviderResult.StopProcessing(
}
}
/*
* If the object has been renamed more recently than this
* operation, cancel the operation.
*/
{
return new SynchronizationProviderResult.StopProcessing(
}
}
else
{
// There is no replication context attached to the operation
// so this is not a replication operation.
{
}
}
return new SynchronizationProviderResult.ContinueProcessing();
}
/**
* 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
*/
{
{
return new SynchronizationProviderResult.StopProcessing(
}
if (fractionalConfig.isFractional())
{
{
/*
* Filter attributes here for fractional replication. If fractional
* replication is enabled, we analyze the operation and modify it so
* backend. This must be called before any other plugin is called, to
* keep coherency across plugin calls.
*/
if (fractionalFilterOperation(modifyOperation, true) ==
{
// Every modifications filtered in this operation: the operation
// becomes a no-op
return new SynchronizationProviderResult.StopProcessing(
}
}
else
{
/*
* Direct access from an LDAP client : if some attributes are to be
* removed according to the fractional configuration, simply forbid
* the operation
*/
switch(fractionalFilterOperation(modifyOperation, false))
{
// Ok, let the operation happen
break;
// Some attributes not compliant with fractional configuration :
// forbid the operation
return new SynchronizationProviderResult.StopProcessing(
}
}
}
{
// No replication ctx attached => not a replicated operation
// - create a ctx with : CSN, entryUUID
// - attach the context to the op
}
else
{
// Replication ctx attached => this is a replicated operation being
// replayed here, 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_OBJECT.
* 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.StopProcessing(
}
// Solve the conflicts between modify operations
{
}
}
return new SynchronizationProviderResult.ContinueProcessing();
}
/**
* 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 UUID is not set when the handleConflict
* phase is called.
*
* @param addOperation The Add Operation.
*/
{
}
/** {@inheritDoc} */
public void publishReplicaOfflineMsg()
{
}
/**
* 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.
{
"replicationCSN", curCSN));
}
{
if (op.isSynchronizationOperation())
{ // Replaying a sync operation
try
{
}
catch (NoSuchElementException e)
{
return;
}
}
else
{
// 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;
}
// If assured replication is configured, this will prepare blocking
// mechanism. If assured replication is disabled, this returns
// immediately
try
{
}
catch (NoSuchElementException e)
{
return;
}
// If assured replication is enabled, this will wait for the matching
// ack or time out. If assured replication is disabled, this returns
// immediately
try
{
} catch (TimeoutException ex)
{
// This exception may only be raised if assured replication is enabled
}
}
// If the operation is a DELETE on the base entry of the suffix
// that is replicated, the generation is now lost because the
// DB is empty. We need to save it again the next time we add an entry.
&& ((PostOperationDeleteOperation) op)
{
generationIdSavedStatus = false;
}
if (!generationIdSavedStatus)
{
}
}
{
// Remove an unsuccessful non-replication operation from the pending
// changes list.
}
}
/**
* Check if the operation that just happened has cleared a conflict :
* Clearing a conflict happens if the operation has free a DN that
* for which an other entry was in conflict.
* Steps:
* - get the DN freed by a DELETE or MODRDN op
* - search for entries put in the conflict space (dn=entryUUID'+'....)
* because the expected DN was not available (ds-sync-conflict=expected DN)
* - retain the entry with the oldest conflict
* - rename this entry with the freedDN as it was expected originally
*/
{
{
// those operations cannot have cleared a conflict
return;
}
{
}
{
}
else
{
return;
}
try
{
}
catch (DirectoryException e)
{
// can not happen?
logger.traceException(e);
return;
}
{
if (entryToRename == null)
{
}
{
// this conflict is older than the previous, keep it.
}
}
if (entryToRename != null)
{
{
}
}
}
/**
* Rename an Entry Using a synchronization, non-replicated operation.
* This method should be used instead of the InternalConnection methods
* when the operation that need to be run must be local only and therefore
* not replicated to the RS.
*
* @param targetDN The DN of the entry to rename.
* @param newRDN The new RDN to be used.
* @param parentDN The parentDN to be used.
* @param markConflict A boolean indicating is this entry should be marked
* as a conflicting entry. In such case the
* DS_SYNC_CONFLICT attribute will be added to the entry
* with the value of its original DN.
* If false, the DS_SYNC_CONFLICT attribute will be
* cleared.
*
* @return The operation that was run to rename the entry.
*/
boolean markConflict)
{
if (markConflict)
{
}
else
{
}
return newOp;
}
{
op.setInternalOperation(true);
op.setSynchronizationOperation(true);
op.setDontSynchronize(true);
}
/**
* Delete this ReplicationDomain.
*/
void delete()
{
shutdown();
}
/**
* Shutdown this ReplicationDomain.
*/
public void shutdown()
{
if (shutdown.compareAndSet(false, true))
{
{
}
// stop the thread in charge of flushing the ServerState.
if (flushThread != null)
{
synchronized (flushThread)
{
}
}
// stop the ReplicationDomain
}
// wait for completion of the ServerStateFlush thread.
try
{
while (!done)
{
}
} catch (InterruptedException e)
{
}
}
/**
* Create and replay a synchronized Operation from an UpdateMsg.
*
* @param msg
* The UpdateMsg to be replayed.
* @param shutdown
* whether the server initiated shutdown
*/
{
// Try replay the operation, then flush (replaying) any pending operation
// whose dependency has been replayed until no more left.
do
{
boolean dependency = false;
try
{
// The next operation for which to attempt replay.
// This local variable allow to keep error messages in the "op" local
// variable until the next loop iteration starts.
// "op" is already initialized to the next Operation because of the
// error handling paths.
boolean replayDone = false;
int retryCount = 10;
{
{
// shutdown initiated, let's leave
return;
}
// Try replay the operation
op.setInternalOperation(true);
op.setSynchronizationOperation(true);
// Always add the ManageDSAIT control so that updates to referrals
// are processed locally.
{
{
// Pre-operation conflict resolution detected that the operation
// was a no-op. For example, an add which has already been
// replayed, or a modify DN operation on an entry which has been
// renamed by a more recent modify DN.
replayDone = true;
}
{
/*
* We probably could not get a lock (OPENDJ-885). Give the server
* another chance to process this operation immediately.
*/
continue;
}
{
/*
* It can happen when a rebuild is performed or the backend is
* offline (OPENDJ-49). Give the server another chance to process
* this operation after some time.
*/
continue;
}
else if (op instanceof ModifyOperation)
{
}
else if (op instanceof DeleteOperation)
{
}
else if (op instanceof AddOperation)
{
}
else if (op instanceof ModifyDNOperation)
{
}
else
{
replayDone = true; // unknown type of operation ?!
}
if (replayDone)
{
// 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
{
/*
* Create a new operation reflecting the new state of the
* UpdateMsg after conflict resolution modified it.
* Note: When msg is a DeleteMsg, the DeleteOperation is properly
* created with subtreeDelete request control when needed.
*/
}
}
else
{
replayDone = true;
}
}
if (!replayDone && !dependency)
{
// Continue with the next change but the servers could now become
// inconsistent.
// Let the repair tool know about this.
}
} catch (DecodeException e)
{
} catch (LDAPException e)
{
} catch (DataFormatException e)
{
} catch (Exception e)
{
{
/*
* 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...
}
{
}
/**
* 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 csn the CSN of the operation with error.
*/
{
try
{
}
catch (NoSuchElementException e)
{
// A failure occurred after the change had been removed from the pending
// changes table.
if (logger.isTraceEnabled())
{
"LDAPReplicationDomain.updateError: Unable to find remote "
+ "pending change for CSN %s", csn);
}
}
}
/**
* Generate a new CSN and insert it in the pending list.
*
* @param operation
* The operation for which the CSN must be generated.
* @return The new CSN.
*/
{
}
/**
* Find the Unique Id of the entry with the provided DN by doing a
* search of the entry and extracting its entryUUID from its attributes.
*
* @param dn The dn of the entry for which the unique Id is searched.
*
* @return The unique Id of the entry with the provided DN.
*/
{
{
return null;
}
if (resultEntry != null)
{
return getEntryUUID(resultEntry);
}
return null;
}
{
{
{
}
}
return null;
}
/**
* find the current DN of an entry from its entry UUID.
*
* @param uuid the Entry Unique ID.
* @return The current DN of the entry or null if there is no entry with
* the specified UUID.
*/
{
try
{
final SearchRequest request = newSearchRequest(getBaseDN(), SearchScope.WHOLE_SUBTREE, "entryuuid=" + uuid);
if (resultEntry != null)
{
return resultEntry.getName();
}
}
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 entry does not exist anymore.
return true;
}
// The modify operation is trying to delete the value that is
// currently used in the RDN. We need to alter the modify so that it does
// not remove the current RDN value(s).
{
{
// the attribute can't be deleted because it is used in the RDN,
// turn this operation is a replace with the current RDN value(s);
}
}
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 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 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.
*/
throws Exception
{
/*
* 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.
*/
// get the current DN of this entry in the database.
// Construct the new DN to use for the entry.
if (newSuperiorID != null)
{
}
else
{
}
//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 targeted 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, 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;
}
{
/*
* 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.
*/
throws Exception
{
{
/*
* 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 log a
* message for the repair tool to look at this problem.
* TODO : Log the message
*/
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.
*/
// reset the parent entryUUID so that the check done is the
// handleConflict phase does not fail.
}
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 entryUUID 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 entryDN The DN of the entry whose child must be renamed.
* @param conflictOp The Operation that generated the conflict.
*/
{
/*
* TODO JNR Ludo thinks that: "Ideally, the operation should verify that the
* entryUUID has not changed or try to use the entryUUID rather than the
* DN.". entryUUID can be obtained from the caller of the current method.
*/
boolean conflict = false;
// Find and rename child entries.
{
{
/*
* Check the ADD and ModRDN date of the child entry
* (All of them, not only the one that are newer than the DEL op)
* and keep the entry as a conflicting entry.
*/
conflict = true;
}
}
else
{
// log error and information for the REPAIR tool.
}
return conflict;
}
/**
* 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 entryUUID 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.
true);
{
// Log information for the repair tool.
}
// Generate an alert to let the administration 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 DecodeException When an encoding error happened manipulating the
* msg.
*/
{
// Generate an alert to let the administrator know that some
// conflict could not be solved.
// Add the conflict attribute
}
/**
* Generate the Dn to use for a conflicting entry.
*
* @param entryUUID 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 entryUUID 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.
*/
{
try
{
} catch (DirectoryException e)
{
// cannot happen
return null;
}
}
/**
* Check if the domain solve conflicts.
*
* @return a boolean indicating if the domain should solve conflicts.
*/
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;
disableService(); // This will cut the session and wake up the listener
}
/**
* Do what necessary when the data have changed : load state, load
* generation Id.
* If there is no such information check if there is a
* ReplicaUpdateVector entry and translate it into a state
* and generationId.
* @exception DirectoryException Thrown when an error occurs.
*/
private void loadDataState() throws DirectoryException
{
// Retrieves the generation ID associated with the data imported
}
/**
* 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 Synchronization
* server.
* The generationId will be retrieved or computed if necessary.
* The ServerState will also be read again from the local database.
*/
public void enable()
{
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;
}
disabled = false;
}
/**
* 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.
*/
private long computeGenerationId() throws DirectoryException
{
if (logger.isTraceEnabled())
{
}
return genId;
}
/**
* Run a modify operation to update the entry whose DN is given as
* a parameter with the generationID information.
*
* @param entryDN The DN of the entry to be updated.
* @param generationId The value of the generationID to be saved.
*
* @return A ResultCode indicating if the operation was successful.
*/
{
// The generationId is stored in the root entry of the domain.
asn1BaseDn, mods);
return op.getResultCode();
}
/**
* Stores the value of the generationId.
* @param generationId The value of the generationId.
* @return a ResultCode indicating if the method was successful.
*/
{
{
generationIdSavedStatus = false;
{
// If the base entry does not exist, save the generation
// ID in the config entry
}
{
}
}
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.
*/
private long loadGenerationId() throws DirectoryException
{
if (logger.isTraceEnabled())
{
}
/*
* Search the database entry that is used to periodically
* save the generation id
*/
{
// if the base entry does not exist look for the generationID
// in the config entry.
}
boolean found = false;
long aGenerationId = -1;
{
{
}
}
else
{
if (resultEntry != null)
{
{
{
String errorMsg = "#Values=" + attr.size() + " Must be exactly 1 in entry " + resultEntry.toLDIFString();
}
{
found = true;
try
{
}
catch(Exception e)
{
}
}
}
}
}
if (!found)
{
if (logger.isTraceEnabled())
{
logger.trace("Generation ID created for domain baseDN=" + getBaseDN() + " generationId=" + aGenerationId);
}
}
else
{
generationIdSavedStatus = true;
if (logger.isTraceEnabled())
{
+ " generationId=" + aGenerationId);
}
}
return aGenerationId;
}
/**
* Do whatever is needed when a backup is started.
* We need to make sure that the serverState is correctly save.
*/
void backupStart()
{
}
/** Do whatever is needed when a backup is finished. */
void backupEnd()
{
// Nothing is needed at the moment
}
/*
* Total Update >>
*/
/**
* This method trigger an export of the replicated data.
*
* @param output The OutputStream where the export should
* be produced.
* @throws DirectoryException When needed.
*/
{
exportBackend(output, false);
}
/**
* The ieContext must have been set before calling.
*
* @param output The OutputStream where the export should
* be produced.
* @param checksumOutput A boolean indicating if this export is
* invoked to perform a checksum only
*
* @return The computed GenerationID.
*
* @throws DirectoryException when an error occurred
*/
throws DirectoryException
{
// Acquire a shared lock for the backend.
try
{
{
}
}
catch (Exception e)
{
}
if (checksumOutput)
{
try
{
}
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 (checksumOutput)
{
{"objectclass", "sn", "cn", "entryuuid"};
{
{
}
}
}
// Launch the export.
long genID = 0;
try
{
}
catch (DirectoryException de)
{
{
}
}
catch (Exception e)
{
LocalizableMessage message = ERR_LDIFEXPORT_ERROR_DURING_EXPORT.get(stackTraceToSingleLineString(e));
}
finally
{
// Clean up after the export by closing the export config.
// Will also flush the export and export the remaining entries.
if (checksumOutput)
{
}
// Release the shared lock on the backend.
try
{
{
}
}
catch (Exception e)
{
}
}
return genID;
}
/**
* Process backend before import.
*
* @param backend
* The backend.
* @throws DirectoryException
* If the backend could not be disabled or locked exclusively.
*/
{
// Stop saving state
stateSavingDisabled = true;
// FIXME setBackendEnabled should be part of TaskUtils ?
// Acquire an exclusive lock for the backend.
{
LocalizableMessage message = ERR_INIT_CANNOT_LOCK_BACKEND.get(backend.getBackendID(), failureReason);
}
}
/**
* This method triggers an import of the replicated data.
*
* @param input The InputStream from which the data are read.
* @throws DirectoryException When needed.
*/
{
try
{
{
return;
}
importConfig.setAppendToExistingData(false);
importConfig.setSkipDNValidation(true);
// We should not validate schema for replication
importConfig.setValidateSchema(false);
// Allow fractional replication ldif import plugin to be called
// Reset the follow import flag and message before starting the import
importErrorMessageId = -1;
// TODO How to deal with rejected entries during the import
// Process import
stateSavingDisabled = false;
}
catch(Exception e)
{
}
finally
{
try
{
// Cleanup
if (importConfig != null)
{
backend = getBackend();
}
{
// When an error occurred during an import, most of times
// the generationId coming in the root entry of the imported data,
// is not valid anymore (partial data in the backend).
}
}
catch (DirectoryException fe)
{
// If we already catch an Exception it's quite possible
// that the loadDataState() and setGenerationId() fail
// so we don't bother about the new Exception.
// However if there was no Exception before we want
// to return this Exception to the task creator.
}
}
{
throw ieCtx.getException();
}
}
/**
* Make post import operations.
* @param backend The backend implied in the import.
* @exception DirectoryException Thrown when an error occurs.
*/
{
// 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))
{
}
// From the domainDN retrieves the replication domain
{
break;
}
if (replicationDomain != null)
{
// Should never happen
}
}
if (replicationDomain == null)
{
}
return replicationDomain;
}
/**
* Returns the backend associated to this domain.
* @return The associated backend.
*/
private Backend<?> getBackend()
{
}
/*
* <<Total Update
*/
/**
* Push the schema modifications contained in the given parameter as a
* modification that would happen on a local server. The modifications are not
* applied to the local schema backend and historical information is not
* updated; but a CSN is generated and the ServerState associated to the
* schema domain is updated.
*
* @param modifications
* The schema modifications to push
*/
{
try
{
}
catch (DirectoryException e)
{
logger.traceException(e);
return;
}
}
/**
* 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 acceptable.
*
* @return true if the configuration is acceptable, false other wise.
*/
{
// 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;
}
// Check fractional configuration
try
{
} catch (ConfigException e)
{
return false;
}
return true;
}
/** {@inheritDoc} */
{
this.config = configuration;
// Read assured + fractional configuration and each time reconnect if needed
readAssuredConfig(configuration, true);
readFractionalConfig(configuration, true);
try
{
}
catch(Exception e)
{
}
return ccr;
}
/** {@inheritDoc} */
public boolean isConfigurationChangeAcceptable(
{
if (ieRunning())
{
return false;
}
// Check fractional configuration
try
{
return true;
}
catch (ConfigException e)
{
return false;
}
}
/** {@inheritDoc} */
{
return alerts;
}
/** {@inheritDoc} */
public String getClassName()
{
return CLASS_NAME;
}
/** {@inheritDoc} */
public DN getComponentEntryDN()
{
}
/**
* Starts the Replication Domain.
*/
public void start()
{
// Create the ServerStateFlush thread
flushThread.start();
}
/**
* Remove from this domain configuration, the configuration of the
* external change log.
*/
private void removeECLDomainCfg()
{
try
{
{
}
}
catch(Exception e)
{
logger.traceException(e);
}
}
/**
* Store the provided ECL configuration for the domain.
* @param domCfg The provided configuration.
* @throws ConfigException When an error occurred.
*/
throws ConfigException
{
// create the ecl config if it does not exist
// There may not be any config entry related to this domain in some
// unit test cases
try
{
{
try
} catch(Exception e) { /* do nothing */ }
// domain with no config entry only when running unit tests
{
// no ECL config provided hence create a default one
// create the default one
{
// no entry exist yet for the ECL config for this domain
// create it
"dn: cn=external changelog," + configDn,
"objectClass: top",
"objectClass: ds-cfg-external-changelog-domain",
"cn: external changelog",
new StringReader(ldif));
// No need to validate schema in replication
ldifImportConfig.setValidateSchema(false);
}
}
}
{
}
else
{
// Create the ECL domain object
}
}
catch (Exception e)
{
}
}
{
}
// Append an extra line so we can append LDIF Strings.
}
/** {@inheritDoc} */
{
// Check domain fractional configuration consistency with local
// configuration variables
// Now for bad data set status if needed
if (forceBadDataSet)
{
return; // Do not send changes to the replication server
}
try
{
/*
* We must not publish changes to a replicationServer that has
* not seen all our previous changes because this could cause
* some other ldap servers to miss those changes.
* Check that the ReplicationServer has seen all our previous
* changes.
*/
// we don't want to update from here (a DS) an empty RS because
// normally the RS should have been updated by other RSes except for
// very last changes lost if the local connection was broken
// ... hence the RS we are connected to should not be empty
// ... or if it is empty, it is due to a voluntary reset
// and we don't want to update it with our changes that could be huge.
{
{
pendingChanges.setRecovering(true);
broker.setRecoveryRequired(true);
{
}
}
}
} catch (Exception e)
{
}
}
/**
* Build the list of changes that have been processed by this server after the
* CSN given as a parameter and publish them using the given session.
*
* @param startCSN
* The CSN where we need to start the search
* @param session
* The session to use to publish the changes
* @return A boolean indicating he success of the operation.
* @throws Exception
* if an Exception happens during the search.
*/
throws Exception
{
// Trim the changes in replayOperations that are older than the startCSN.
synchronized (replayOperations)
{
{
{
return false;
}
{
break;
}
}
}
do
{
{
return false;
}
// We can't do the search in one go because we need to store the results
// so that we are sure we send the operations in order and because the
// list might be large.
// So we search by interval of 10 seconds and store the results in the
// replayOperations list so that they are sorted before sending them.
listener);
// Publish and remove all the changes from the replayOperations list
// that are older than the endCSN.
synchronized (replayOperations)
{
{
{
return false;
}
// do not look for replay operations in the future
{
break;
}
}
}
{
{
return false;
}
}
if (lastRetrievedChange != null)
{
if (logger.isInfoEnabled())
+ " lastRetrievedChange=" + lastRetrievedChange));
}
else
{
if (logger.isInfoEnabled())
+ " no changes"));
}
}
{
// ensure now() will always come last with isNewerThan() test,
// even though the timestamp, or the timestamp and seqnum would be the same
}
/**
* Search for the changes that happened since fromCSN based on the historical
* attribute. The only changes that will be send will be the one generated on
* the serverId provided in fromCSN.
*
* @param baseDN
* the base DN
* @param fromCSN
* The CSN from which we want the changes
* @param lastCSN
* The max CSN that the search should return
* @param resultListener
* The listener that will process the entries returned
* @return the internal search operation
* @throws Exception
* when raised.
*/
throws Exception
{
{
+ "ffffffff";
}
else
{
}
}
/**
* Search for the changes that happened since fromCSN based on the historical
* attribute. The only changes that will be send will be the one generated on
* the serverId provided in fromCSN.
*
* @param baseDN
* the base DN
* @param fromCSN
* The CSN from which we want the changes
* @param resultListener
* that will process the entries returned.
* @return the internal search operation
* @throws Exception
* when raised.
*/
{
}
/**
* This method should return the total number of objects in the
* replicated domain.
* This count will be used for reporting.
*
* @throws DirectoryException when needed.
*
* @return The number of objects in the replication domain.
*/
public long countEntries() throws DirectoryException
{
{
}
}
/** {@inheritDoc} */
{
// Ignore message if fractional configuration is inconsistent and
// we have been passed into bad data set status
if (forceBadDataSet)
{
return false;
}
if (updateMsg instanceof LDAPUpdateMsg)
{
// Put the UpdateMsg in the RemotePendingChanges list.
{
/*
* Already received this change so ignore it. This may happen if there
* are uncommitted changes in the queue and session failover occurs
* causing a recovery of all changes since the current committed server
* state. See OPENDJ-1115.
*/
if (logger.isTraceEnabled())
{
"LDAPReplicationDomain.processUpdate: ignoring "
}
return true;
}
// Put update message into the replay queue
// (block until some place in the queue is available)
while (!isListenerShuttingDown())
{
// loop until we can offer to the queue or shutdown was initiated
try
{
{
// successful offer to the queue, let's exit the loop
break;
}
}
catch (InterruptedException e)
{
// Thread interrupted: check for shutdown.
}
}
return false;
}
// unknown message type, this should not happen, just ignore it.
return true;
}
/**
* Monitoring information for the LDAPReplicationDomain.
*
* @return Monitoring attributes specific to the LDAPReplicationDomain.
*/
{
// number of updates in the pending list
return attributes;
}
/**
* 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 integer value
* @throws DirectoryException if the string is not valid
*/
{
int source = 0;
try
{
{
// TODO Verifies serverID is in the domain
// We should check here that this is a server implied
// in the current domain.
return source;
}
}
catch (Exception e)
{
}
}
/**
* Called by synchronize post op plugin in order to add the entry historical
* attributes to the UpdateMsg.
* @param msg an replication update message
* @param op the operation in progress
*/
{
if (op instanceof PostOperationDeleteOperation)
{
// For delete only, add the Authorized DN since it's required in the
// ECL entry but is not part of rest of the message.
{
}
}
else if (op instanceof PostOperationModifyOperation)
{
}
else if (op instanceof PostOperationModifyDNOperation)
{
}
else if (op instanceof PostOperationAddOperation)
{
}
}
{
{
// Fast-path.
return Collections.emptySet();
}
{
// Potential fast-path for delete operations.
{
}
if (objectClassAttribute != null)
{
}
return attributes;
}
else
{
// Expand @objectclass references in attribute list if needed.
// We do this now in order to take into account dynamic schema changes.
final Entry filteredEntry =
return filteredEntry.getAttributes();
}
}
{
// Only rebuild the attribute set if necessary.
if (!needsExpanding(names))
{
return names;
}
{
{
if (objectClass != null)
{
{
}
{
}
}
}
else
{
}
}
return expandedNames;
}
{
{
{
return true;
}
}
return false;
}
/**
* Gets the fractional configuration of this domain.
* @return The fractional configuration of this domain.
*/
{
return fractionalConfig;
}
/**
* This bean is a utility class used for holding the parsing
* result of a fractional configuration. It also contains some facility
* methods like fractional configuration comparison...
*/
static class FractionalConfig
{
/**
* Tells if fractional replication is enabled or not (some fractional
* constraints have been put in place). If this is true then
* fractionalExclusive explains the configuration mode and either
* fractionalSpecificClassesAttributes or fractionalAllClassesAttributes or
* both should be filled with something.
*/
private boolean fractional;
/**
* - If true, tells that the configured fractional replication is exclusive:
* Every attributes contained in fractionalSpecificClassesAttributes and
* fractionalAllClassesAttributes should be ignored when replaying operation
* in local backend.
* - If false, tells that the configured fractional replication is
* inclusive:
* Only attributes contained in fractionalSpecificClassesAttributes and
* fractionalAllClassesAttributes should be taken into account in local
* backend.
*/
private boolean fractionalExclusive = true;
/**
* Used in fractional replication: holds attributes of a specific object
* class.
* - key = object class (name or OID of the class)
* - value = the attributes of that class that should be taken into account
* (inclusive or exclusive fractional replication) (name or OID of the
* attribute)
* When an operation coming from the network is to be locally replayed, if
* the concerned entry has an objectClass attribute equals to 'key':
* modified
* modified
*/
/**
* Used in fractional replication: holds attributes of any object class.
* When an operation coming from the network is to be locally replayed:
* - inclusive mode: only attributes of the matching entry not present in
* - exclusive mode: attributes of the matching entry present in
* The attributes may be in human readable form of OID form.
*/
/**
* Base DN the fractional configuration is for.
*/
/**
* Constructs a new fractional configuration object.
* @param baseDN The base DN the object is for.
*/
{
}
/**
* Getter for fractional.
* @return True if the configuration has fractional enabled
*/
boolean isFractional()
{
return fractional;
}
/**
* Set the fractional parameter.
* @param fractional The fractional parameter
*/
private void setFractional(boolean fractional)
{
this.fractional = fractional;
}
/**
* Getter for fractionalExclusive.
* @return True if the configuration has fractional exclusive enabled
*/
boolean isFractionalExclusive()
{
return fractionalExclusive;
}
/**
* Set the fractionalExclusive parameter.
* @param fractionalExclusive The fractionalExclusive parameter
*/
private void setFractionalExclusive(boolean fractionalExclusive)
{
}
/**
* Getter for fractionalSpecificClassesAttributes attribute.
* @return The fractionalSpecificClassesAttributes attribute.
*/
{
}
/**
* Set the fractionalSpecificClassesAttributes parameter.
* @param fractionalSpecificClassesAttributes The
* fractionalSpecificClassesAttributes parameter to set.
*/
private void setFractionalSpecificClassesAttributes(
{
}
/**
* Getter for fractionalSpecificClassesAttributes attribute.
* @return The fractionalSpecificClassesAttributes attribute.
*/
{
return fractionalAllClassesAttributes;
}
/**
* Set the fractionalAllClassesAttributes parameter.
* @param fractionalAllClassesAttributes The
* fractionalSpecificClassesAttributes parameter to set.
*/
private void setFractionalAllClassesAttributes(
{
}
/**
* Getter for the base baseDN.
* @return The baseDN attribute.
*/
{
return baseDN;
}
/**
* Extract the fractional configuration from the passed domain configuration
* entry.
* @param configuration The configuration object
* @return The fractional replication configuration.
* @throws ConfigException If an error occurred.
*/
static FractionalConfig toFractionalConfig(
{
// Prepare fractional configuration variables to parse
// Get potentially new fractional configuration
// Create matching parsed config object
switch (newFractionalMode)
{
case NOT_FRACTIONAL:
result.setFractional(false);
result.setFractionalExclusive(true);
break;
case EXCLUSIVE_FRACTIONAL:
case INCLUSIVE_FRACTIONAL:
result.setFractional(true);
break;
}
return result;
}
/**
* Parses a fractional replication configuration, filling the empty passed
* variables and returning the used fractional mode. The 2 passed variables
* to fill should be initialized (not null) and empty.
* @param exclIt The list of fractional exclude configuration values (may be
* null)
* @param inclIt The list of fractional include configuration values (may be
* null)
* @param fractionalSpecificClassesAttributes An empty map to be filled with
* what is read from the fractional configuration properties.
* @param fractionalAllClassesAttributes An empty list to be filled with
* what is read from the fractional configuration properties.
* @return the fractional mode deduced from the passed configuration:
* not fractional, exclusive fractional or inclusive fractional
* modes
*/
private static int parseFractionalConfig (
{
// Determine if fractional-exclude or fractional-include property is used:
// only one of them is allowed
int fractionalMode;
{
{
throw new ConfigException(
}
}
else
{
{
}
else
{
return NOT_FRACTIONAL;
}
}
{
// Parse a value with the form class:attr1,attr2...
// or *:attr1,attr2...
if (nTokens < 2)
{
get(fractCfgStr));
}
// Get the class name
// Get the attributes
while (st.hasMoreTokens())
{
// Store attribute in the appropriate variable
if (allClasses)
{
}
else
{
{
}
}
}
}
return fractionalMode;
}
/** Return type of the parseFractionalConfig method. */
private static final int NOT_FRACTIONAL = 0;
private static final int EXCLUSIVE_FRACTIONAL = 1;
private static final int INCLUSIVE_FRACTIONAL = 2;
/**
* Get an integer representation of the domain fractional configuration.
* @return An integer representation of the domain fractional configuration.
*/
private int fractionalConfigToInt()
{
if (!fractional)
{
return NOT_FRACTIONAL;
}
else if (fractionalExclusive)
{
return EXCLUSIVE_FRACTIONAL;
}
return INCLUSIVE_FRACTIONAL;
}
/**
* Compare 2 fractional replication configurations and returns true if they
* are equivalent.
* @param cfg1 First fractional configuration
* @param cfg2 Second fractional configuration
* @return True if both configurations are equivalent.
* @throws ConfigException If some classes or attributes could not be
* retrieved from the schema.
*/
{
// Compare base DNs just to be consistent
{
return false;
}
// Compare modes
{
return false;
}
// Compare all classes attributes
{
return false;
}
// Compare specific classes attributes
{
return false;
}
/*
* Check consistency of specific classes attributes
*
* For each class in specificClassesAttributes1, check that the attribute
* list is equivalent to specificClassesAttributes2 attribute list
*/
{
// Get class from specificClassesAttributes1
if (objectClass1 == null)
{
throw new ConfigException(
}
// Look for matching one in specificClassesAttributes2
boolean foundClass = false;
{
if (objectClass2 == null)
{
throw new ConfigException(
}
{
foundClass = true;
// Now compare the 2 attribute lists
{
return false;
}
break;
}
}
// Found matching class ?
if (!foundClass)
{
return false;
}
}
return true;
}
}
/**
*/
boolean isECLEnabled()
{
}
/**
* Return the minimum time (in ms) that the domain keeps the historical
* information necessary to solve conflicts.
*
* @return the purge delay.
*/
long getHistoricalPurgeDelay()
{
}
/**
* Check if the operation that just happened has cleared a conflict : Clearing
* a conflict happens if the operation has freed a DN for which another entry
* was in conflict.
* <p>
* Steps:
* <ul>
* <li>get the DN freed by a DELETE or MODRDN op</li>
* <li>search for entries put in the conflict space (dn=entryUUID'+'....)
* because the expected DN was not available (ds-sync-conflict=expected DN)
* </li>
* <li>retain the entry with the oldest conflict</li>
* <li>rename this entry with the freedDN as it was expected originally</li>
* </ul>
*
* @param task
* the task raising this purge.
* @param endDate
* the date to stop this task whether the job is done or not.
* @throws DirectoryException
* when an exception happens.
*/
long endDate) throws DirectoryException
{
+ "on domain: " + getBaseDN()
+ "lastCSNPurgedFromHist: "
int count = 0;
{
}
{
if (maxTimeToRun < 0)
{
}
{
// Log information for the repair tool.
}
{
}
}
}
}