ReplicationDomain.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 2008-2010 Sun Microsystems, Inc.
* Portions Copyright 2011-2015 ForgeRock AS
*/
/**
* This class should be used as a base for Replication implementations.
* <p>
* It is intended that developer in need of a replication mechanism
* subclass this class with their own implementation.
* <p>
* The startup phase of the ReplicationDomain subclass,
* should read the list of replication servers from the configuration,
* instantiate a {@link ServerState} then start the publish service
* by calling {@link #startPublishService()}.
* At this point it can start calling the {@link #publish(UpdateMsg)}
* method if needed.
* <p>
* When the startup phase reach the point when the subclass is ready
* to handle updates the Replication Domain implementation should call the
* {@link #startListenService()} method.
* At this point a Listener thread is created on the Replication Service
* and which can start receiving updates.
* <p>
* When updates are received the Replication Service calls the
* {@link #processUpdate(UpdateMsg)} method.
* ReplicationDomain implementation should implement the appropriate code
* for replaying the update on the local repository.
* When fully done the subclass must call the
* {@link #processUpdateDone(UpdateMsg, String)} method.
* This allows to process the update asynchronously if necessary.
*
* <p>
* To propagate changes to other replica, a ReplicationDomain implementation
* must use the {@link #publish(UpdateMsg)} method.
* <p>
* If the Full Initialization process is needed then implementation
* for {@code importBackend(InputStream)} and
* {@code exportBackend(OutputStream)} must be
* provided.
* <p>
* Full Initialization of a replica can be triggered by LDAP clients
* by creating InitializeTasks or InitializeTargetTask.
* Full initialization can also be triggered from the ReplicationDomain
* implementation using methods {@link #initializeRemote(int, Task)}
* or {@link #initializeFromRemote(int, Task)}.
* <p>
* At shutdown time, the {@link #disableService()} method should be called to
* cleanly stop the replication service.
*/
public abstract class ReplicationDomain
{
/**
* Contains all the attributes included for the ECL (External Changelog).
*/
// @Immutable
private final static class ECLIncludes
{
private ECLIncludes(
{
}
@SuppressWarnings("unchecked")
public ECLIncludes()
{
}
/**
* Add attributes to be included in the ECL.
*
* @param serverId
* Server where these attributes are configured.
* @param includeAttributes
* Attributes to be included with all change records, may include
* wild-cards.
* @param includeAttributesForDeletes
* Additional attributes to be included with delete change records,
* may include wild-cards.
* @return a new {@link ECLIncludes} object if included attributes have
* changed, or the current object otherwise.
*/
{
boolean configurationChanged = false;
// Combine all+delete attributes.
{
configurationChanged = true;
this.includedAttrsByServer);
}
{
configurationChanged = true;
}
if (!configurationChanged)
{
return this;
}
// and rebuild the global list to be ready for usage
{
}
{
}
return new ECLIncludes(eclIncludesByServer,
}
}
/**
* Current status for this replicated domain.
*/
/** The configuration of the replication domain. */
protected volatile ReplicationDomainCfg config;
/**
* The assured configuration of the replication domain. It is a duplicate of
* {@link #config} because of its update model.
*
* @see #readAssuredConfig(ReplicationDomainCfg, boolean)
*/
private volatile ReplicationDomainCfg assuredConfig;
/**
* The ReplicationBroker that is used by this ReplicationDomain to
* connect to the ReplicationService.
*/
protected ReplicationBroker broker;
/**
* This Map is used to store all outgoing assured messages in order
* to be able to correlate all the coming back acks to the original
* operation.
*/
/**
* The context related to an import or export being processed
* Null when none is being processed.
*/
new AtomicReference<ImportExportContext>();
/**
* The 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 volatile DirectoryThread listenerThread;
/**
* A set of counters used for Monitoring.
*/
/** Assured replication monitoring counters. */
/** Number of updates sent in Assured Mode, Safe Read. */
/**
* Number of updates sent in Assured Mode, Safe Read, that have been
* successfully acknowledged.
*/
/**
* Number of updates sent in Assured Mode, Safe Read, that have not been
* successfully acknowledged (either because of timeout, wrong status or error
* at replay).
*/
private AtomicInteger assuredSrNotAcknowledgedUpdates =
new AtomicInteger(0);
/**
* Number of updates sent in Assured Mode, Safe Read, that have not been
* successfully acknowledged because of timeout.
*/
/**
* Number of updates sent in Assured Mode, Safe Read, that have not been
* successfully acknowledged because of wrong status.
*/
/**
* Number of updates sent in Assured Mode, Safe Read, that have not been
* successfully acknowledged because of replay error.
*/
/**
* Multiple values allowed: number of updates sent in Assured Mode, Safe Read,
* that have not been successfully acknowledged (either because of timeout,
* wrong status or error at replay) for a particular server (DS or RS).
* <p>
* String format: <server id>:<number of failed updates>
*/
/** Number of updates received in Assured Mode, Safe Read request. */
/**
* Number of updates received in Assured Mode, Safe Read request that we have
* acked without errors.
*/
/**
* Number of updates received in Assured Mode, Safe Read request that we have
* acked with errors.
*/
/** Number of updates sent in Assured Mode, Safe Data. */
/**
* Number of updates sent in Assured Mode, Safe Data, that have been
* successfully acknowledged.
*/
/**
* Number of updates sent in Assured Mode, Safe Data, that have not been
* successfully acknowledged because of timeout.
*/
/**
* Multiple values allowed: number of updates sent in Assured Mode, Safe Data,
* that have not been successfully acknowledged because of timeout for a
* particular RS.
* <p>
* String format: <server id>:<number of failed updates>
*/
/* Status related monitoring fields */
/**
* Indicates the date when the status changed. This may be used to indicate
* the date the session with the current replication server started (when
* status is NORMAL for instance). All the above assured monitoring fields
* are also reset each time the status is changed
*/
/**
* The state maintained by the Concrete Class.
*/
private final ServerState state;
/**
* The generator that will be used to generate {@link CSN}
* for this domain.
*/
private final CSNGenerator generator;
/**
* An object used to protect the initialization of the underlying broker
* session of this ReplicationDomain.
*/
/**
* The generationId for this replication domain. It is made of a hash of the
* 1000 first entries for this domain.
*/
protected volatile long generationId;
/**
* Returns the {@link CSNGenerator} that will be used to
* generate {@link CSN} for this domain.
*
* @return The {@link CSNGenerator} that will be used to
* generate {@link CSN} for this domain.
*/
public CSNGenerator getGenerator()
{
return generator;
}
/**
* Creates a ReplicationDomain with the provided parameters.
*
* @param config
* The configuration object for this ReplicationDomain
* @param generationId
* the generation of this ReplicationDomain
*/
{
}
/**
* Creates a ReplicationDomain with the provided parameters. (for unit test
* purpose only)
*
* @param config
* The configuration object for this ReplicationDomain
* @param generationId
* the generation of this ReplicationDomain
* @param serverState
* The serverState to use
*/
{
this.assuredConfig = config;
this.generationId = generationId;
this.state = serverState;
}
/**
* Set the initial status of the domain and perform necessary initializations.
* This method will be called by the Broker each time the ReplicationBroker
* establish a new session to a Replication Server.
*
* Implementations may override this method when they need to perform
* additional computing after session establishment.
* The default implementation should be sufficient for ReplicationDomains
* that don't need to perform additional computing.
*
* @param initStatus The status to enter the state machine with.
* @param rsState The ServerState of the ReplicationServer
* with which the session was established.
*/
{
// Sanity check: is it a valid initial status?
if (!isValidInitialStatus(initStatus))
{
}
else
{
status = initStatus;
}
}
/**
* Processes an incoming ChangeStatusMsg. Compute new status according to
* given order. Then update domain for being compliant with new status
* definition.
* @param csMsg The received status message
*/
{
if (logger.isTraceEnabled())
" received change status message:\n" + csMsg);
// Translate requested status to a state machine event
{
return;
}
// Set the new status to the requested one
}
/**
* Called when first connection or disconnection detected.
*/
void toNotConnectedStatus()
{
// Go into not connected status
}
/**
* Perform whatever actions are needed to apply properties for being
* compliant with new status. Must be called in synchronized section for
* status. The new status is already set in status variable.
*/
private void updateDomainForNewStatus()
{
switch (status)
{
case FULL_UPDATE_STATUS:
// Signal RS we just entered the full update status
break;
case NOT_CONNECTED_STATUS:
case NORMAL_STATUS:
case DEGRADED_STATUS:
case BAD_GEN_ID_STATUS:
break;
default:
if (logger.isTraceEnabled())
status);
}
}
/**
* Gets the status for this domain.
* @return The status for this domain.
*/
public ServerStatus getStatus()
{
return status;
}
/**
* Returns the base DN of this ReplicationDomain. All Replication Domain using
* this baseDN will be connected through the Replication Service.
*
* @return The base DN of this ReplicationDomain
*/
{
}
/**
* Get the server ID. The identifier of this Replication Domain inside the
* Replication Service. Each Domain must use a unique ServerID.
*
* @return The server ID.
*/
public int getServerId()
{
return config.getServerId();
}
/**
* Window size used during initialization .. between - the
* initializer/exporter DS that listens/waits acknowledges and that slows down
* data msg publishing based on the slowest server - and each
* initialized/importer DS that publishes acknowledges each WINDOW/2 data msg
* received.
*
* @return the initWindow
*/
protected int getInitWindow()
{
return config.getInitializationWindowSize();
}
/**
* Tells if assured replication is enabled for this domain.
* @return True if assured replication is enabled for this domain.
*/
public boolean isAssured()
{
}
/**
* Gives the mode for the assured replication of the domain. Only used when
* assured is true).
*
* @return The mode for the assured replication of the domain.
*/
public AssuredMode getAssuredMode()
{
switch (assuredConfig.getAssuredType())
{
case SAFE_DATA:
case NOT_ASSURED: // The assured mode will be ignored in that case anyway
return AssuredMode.SAFE_DATA_MODE;
case SAFE_READ:
return AssuredMode.SAFE_READ_MODE;
}
return null; // should never happen
}
/**
* Gives the assured Safe Data level of the replication of the domain. (used
* when assuredMode is SAFE_DATA).
*
* @return The assured level of the replication of the domain.
*/
public byte getAssuredSdLevel()
{
return (byte) assuredConfig.getAssuredSdLevel();
}
/**
* Gives the assured timeout of the replication of the domain (in ms).
* @return The assured timeout of the replication of the domain.
*/
public long getAssuredTimeout()
{
return assuredConfig.getAssuredTimeout();
}
/**
* Gets the group id for this domain.
* @return The group id for this domain.
*/
public byte getGroupId()
{
return (byte) config.getGroupId();
}
/**
* Gets the referrals URLs this domain publishes. Referrals urls to be
* published to other servers of the topology.
* <p>
* TODO: fill that with all currently opened urls if no urls configured
*
* @return The referrals URLs this domain publishes.
*/
{
return config.getReferralsUrl();
}
/**
* Gets the info for Replicas in the topology (except us).
* @return The info for Replicas in the topology (except us)
*/
{
return broker.getReplicaInfos();
}
/**
* Returns information about the DS server related to the provided serverId.
* based on the TopologyMsg we received when the remote replica connected or
* disconnected. Return null when no server with the provided serverId is
* connected.
*
* @param dsId The provided serverId of the remote replica
* @return the info related to this remote server if it is connected,
* null is the server is NOT connected.
*/
{
}
/**
* Gets the States of all the Replicas currently in the
* Topology.
* When this method is called, a Monitoring message will be sent
* to the Replication Server to which this domain is currently connected
* so that it computes a table containing information about
* all Directory Servers in the topology.
* This Computation involves communications will all the servers
* currently connected and
*
* @return The States of all Replicas in the topology (except us)
*/
{
return broker.getReplicaStates();
}
/**
* Gets the info for RSs in the topology (except the one we are connected
* to).
* @return The info for RSs in the topology (except the one we are connected
* to)
*/
{
return broker.getRsInfos();
}
/**
* Gets the server ID of the Replication Server to which the domain
* is currently connected.
*
* @return The server ID of the Replication Server to which the domain
* is currently connected.
*/
public int getRsServerId()
{
return broker.getRsServerId();
}
/**
* Increment the number of processed updates.
*/
private void incProcessedUpdates()
{
}
/**
* get the number of updates replayed by the replication.
*
* @return The number of updates replayed by the replication
*/
int getNumProcessedUpdates()
{
if (numProcessedUpdates != null)
{
return numProcessedUpdates.get();
}
return 0;
}
/**
* get the number of updates received by the replication plugin.
*
* @return the number of updates received
*/
int getNumRcvdUpdates()
{
if (numRcvdUpdates != null)
{
return numRcvdUpdates.get();
}
return 0;
}
/**
* Get the number of updates sent by the replication plugin.
*
* @return the number of updates sent
*/
int getNumSentUpdates()
{
if (numSentUpdates != null)
{
return numSentUpdates.get();
}
return 0;
}
/**
* Receives an update message from the replicationServer.
* The other types of messages are processed in an opaque way for the caller.
* Also responsible for updating the list of pending changes
* @return the received message - null if none
*/
{
{
try
{
{
// The server is in the shutdown process
return null;
}
{
}
{
}
else if (msg instanceof InitializeRequestMsg)
{
// Another server requests us to provide entries
// for a total update
}
else if (msg instanceof InitializeTargetMsg)
{
// Another server is exporting its entries to us
/*
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.
*/
}
{
{
/*
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 the import source.
A remote error during the import will be received in the
receiveEntryBytes() method.
*/
if (logger.isTraceEnabled())
"[IE] processErrorMsg:" + getServerId() +
" baseDN: " + getBaseDN() +
" Error Msg received: " + errorMsg);
{
}
else
{
/*
Simply log - happen when the ErrorMsg relates to a previous
attempt of initialization while we have started a new one
on this side.
*/
}
}
else
{
// on our side before receiving this ErrorMsg.
}
}
else if (msg instanceof ChangeStatusMsg)
{
}
{
}
else if (msg instanceof InitializeRcvAckMsg)
{
{
}
}
}
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 receiving 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.
*/
if (initReqMsg != null)
{
// Do this work in a thread to allow replay thread continue working
}
}
{
}
return update;
}
/**
* Updates the passed monitoring list of errors received for assured messages
* (safe data or safe read, depending of the passed list to update) for a
* particular server in the list. This increments the counter of error for the
* passed server, or creates an initial value of 1 error for it if the server
* is not yet present in the map.
* @param errorsByServer map of number of errors per serverID
* @param sid the ID of the server which produced an error
*/
{
synchronized (errorsByServer)
{
if (serverErrCount == null)
{
// Server not present in list, create an entry with an
// initial number of errors set to 1
} else
{
// Server already present in list, just increment number of
// errors for the server
int val = serverErrCount;
val++;
}
}
}
/**
* Do the necessary processing when an AckMsg is received.
*
* @param ack The AckMsg that was received.
*/
{
// Remove the message for pending ack list (this may already make the thread
// that is waiting for the ack be aware of its reception)
// Signal waiting thread ack has been received
{
synchronized (update)
{
}
// Analyze status of embedded in the ack to see if everything went well
{
/*
Some problems detected: message did not correctly reach every
requested servers. Log problem
*/
// Increment assured replication monitoring counters
switch (updateAssuredMode)
{
case SAFE_READ_MODE:
if (hasTimeout)
{
}
if (hasReplayErrors)
{
}
if (hasWrongStatus)
{
}
{
{
}
}
break;
case SAFE_DATA_MODE:
// The only possible cause of ack error in safe data mode is timeout
if (hasTimeout) // So should always be the case
{
}
{
{
}
}
break;
default:
// Should not happen
}
} else
{
// Update has been acknowledged without errors
// Increment assured replication monitoring counters
switch (updateAssuredMode)
{
case SAFE_READ_MODE:
break;
case SAFE_DATA_MODE:
break;
default:
// Should not happen
}
}
}
}
/*
* After this point the code is related to the Total Update.
*/
/**
* This thread is launched when we want to export data to another server.
*
* When a task is created locally (so this local server is the initiator)
* of the export (Example: dsreplication initialize-all),
* this thread is NOT used but the task thread is running the export instead).
*/
private class ExportThread extends DirectoryThread
{
/** Id of server that will be initialized. */
private final int serverIdToInitialize;
private final int initWindow;
/**
* Constructor for the ExportThread.
*
* @param serverIdToInitialize
* serverId of server that will receive entries
* @param initWindow
* The value of the initialization window for flow control between
* the importer and the exporter.
*/
{
this.initWindow = initWindow;
}
/**
* Run method for this class.
*/
public void run()
{
if (logger.isTraceEnabled())
try
{
} catch (DirectoryException de)
{
/*
An error message has been sent to the peer
This server is not the initiator of the export so there is
nothing more to do locally.
*/
}
if (logger.isTraceEnabled())
}
}
/**
* This class contains the context related to an import or export launched on
* the domain.
*/
protected class ImportExportContext
{
/** The private task that initiated the operation. */
private Task initializeTask;
/** The destination in the case of an export. */
/** The source in the case of an import. */
/** The total entry count expected to be processed. */
private long entryCount;
/** The count for the entry not yet processed. */
private long entryLeftCount;
/** Exception raised during the initialization. */
private DirectoryException exception;
/** Whether the context is related to an import or an export. */
private final boolean importInProgress;
/** Current counter of messages exchanged during the initialization. */
private int msgCnt;
/**
* Number of connections lost when we start the initialization. Will help
* counting connections lost during initialization,
*/
private int initNumLostConnections;
/**
* Request message sent when this server has the initializeFromRemote task.
*/
private InitializeRequestMsg initReqMsgSent;
/**
* Start time of the initialization process. ErrorMsg timestamped before
* this startTime will be ignored.
*/
private final long startTime;
/**
* List for replicas (DS) connected to the topology when initialization
* started.
*/
/**
* List for replicas (DS) with a failure (disconnected from the topology)
* since the initialization started.
*/
/**
* Flow control during initialization: for each remote server, counter of
* messages received.
*/
/**
* ServerId of the slowest server (the one with the smallest non null
* counter).
*/
private int slowestServerId = -1;
private short exporterProtocolVersion = -1;
/** Window used during this initialization. */
private int initWindow;
/** Number of attempt already done for this initialization. */
private short attemptCnt;
/**
* Creates a new IEContext.
*
* @param importInProgress true if the IEContext will be used
* for and import, false if the IEContext
* will be used for and export.
*/
private ImportExportContext(boolean importInProgress)
{
this.importInProgress = importInProgress;
this.attemptCnt = 0;
}
/**
* Returns a boolean indicating if a total update import is currently in
* Progress.
*
* @return A boolean indicating if a total update import is currently in
* Progress.
*/
boolean importInProgress()
{
return importInProgress;
}
/**
* Returns the total number of entries to be processed when a total update
* is in progress.
*
* @return The total number of entries to be processed when a total update
* is in progress.
*/
long getTotalEntryCount()
{
return entryCount;
}
/**
* Returns the number of entries still to be processed when a total update
* is in progress.
*
* @return The number of entries still to be processed when a total update
* is in progress.
*/
long getLeftEntryCount()
{
return entryLeftCount;
}
/**
* @param total Total number of entries to be processed.
* @throws DirectoryException if an error occurred.
*/
{
entryCount = total;
if (initializeTask instanceof InitializeTask)
{
}
else if (initializeTask instanceof InitializeTargetTask)
{
}
}
/**
* Update the counters of the task for each entry processed during
* an import or export.
*
* @param entriesDone The number of entries that were processed
* since the last time this method was called.
*
* @throws DirectoryException if an error occurred.
*/
{
if (initializeTask != null)
{
if (initializeTask instanceof InitializeTask)
{
}
else if (initializeTask instanceof InitializeTargetTask)
{
}
}
}
/** {@inheritDoc} */
{
return "[Entry count=" + this.entryCount +
}
/**
* Gets the server id of the exporting server.
* @return the server id of the exporting server.
*/
public int getExportTarget()
{
return exportTarget;
}
/**
* Gets the server id of the importing server.
* @return the server id of the importing server.
*/
public int getImportSource()
{
return importSource;
}
/**
*/
public DirectoryException getException()
{
return exception;
}
/**
*/
{
}
/**
* was already set on this object.
*
*/
{
{
}
}
/**
* Set the id of the EntryMsg acknowledged from a receiver (importer)server.
* (updated via the listener thread)
* @param serverId serverId of the acknowledger/receiver/importer server.
* @param numAck id of the message received.
*/
{
if (logger.isTraceEnabled())
// Recompute the server with the minAck returned,means the slowest server.
{
{
}
}
}
/**
* Returns the serverId of the server that acknowledged the smallest
* EntryMsg id.
* @return serverId of the server with latest acknowledge.
* 0 when no ack has been received yet.
*/
public int getSlowestServer()
{
if (logger.isTraceEnabled())
return this.slowestServerId;
}
}
/**
* 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 integer value
* @throws DirectoryException if the string is not valid
*/
{
{
return RoutableMsg.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 ?
// JNR: Yes please
}
return target;
}
catch (Exception e)
{
}
}
/**
* Initializes a remote server from this server.
* <p>
* The {@code exportBackend(OutputStream)} will therefore be called
* on this server, and the {@code importBackend(InputStream)}
* will be called on the remote server.
* <p>
* The InputStream and OutputStream given as a parameter to those
* methods will be connected through the replication protocol.
*
* @param target The server-id of the server that should be initialized.
* The target can be discovered using the
* {@link #getReplicaInfos()} method.
* @param initTask The task that triggers this initialization and that should
* be updated with its progress.
*
* @throws DirectoryException If it was not possible to publish the
* Initialization message to the Topology.
*/
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 serverToInitialize The target server that should be initialized.
* @param serverRunningTheTask The server that initiated the export. It can
* be the serverID of this server, or the serverID of a remote server.
* @param initTask The task in this server that triggers this initialization
* and that should be updated with its progress. Null when the export is done
* following a request coming from a remote server (task is remote).
* @param initWindow The value of the initialization window for flow control
* between the importer and the exporter.
*
* @exception DirectoryException When an error occurs. No exception raised
* means success.
*/
protected void initializeRemote(int serverToInitialize,
throws DirectoryException
{
/*
We manage the list of servers to initialize in order :
- to test at the end that all expected servers have reconnected
after their import and with the right genId
- to update the task with the server(s) where this test failed
*/
{
// We manage the list of servers with which a flow control can be enabled
{
{
}
}
}
else
{
// We manage the list of servers with which a flow control can be enabled
{
{
}
}
}
// loop for the case where the exporter is the initiator
int attempt = 0;
boolean done = false;
{
try
{
{
}
// Send start message to the peer
// Wait for all servers to be ok
// Servers that left in the list are those for which we could not test
// that they have been successfully initialized.
{
throw new DirectoryException(
}
// Notify the peer of the success
}
{
// Give priority to the first exception raised - stored in the context
}
if (logger.isTraceEnabled())
+ " exportRootException=" + exportRootException);
if (exportRootException != null)
{
try
{
/*
Handling the errors during export
Note: we could have lost the connection and another thread
the listener one) has already managed to reconnect.
So we MUST rely on the test broker.isConnected()
ONLY to do 'wait to be reconnected by another thread'
(if not yet reconnected already).
*/
if (!broker.isConnected())
{
// We are still disconnected, so we wait for the listener thread
// to reconnect - wait 10s
if (logger.isTraceEnabled())
"[IE] Exporter wait for reconnection by the listener thread");
int att=0;
while (!broker.shuttingDown()
&& !broker.isConnected()
&& ++att < 100)
{
catch(Exception e){ /* do nothing */ }
}
}
&& broker.isConnected()
{
/*
NewAttempt case : In the case where
- it's not an InitializeAll
- AND the previous export attempt failed
- AND we are (now) connected
- and we own the task and this task is not an InitializeAll
Let's :
- sleep to let time to the other peer to reconnect if needed
- and launch another attempt
*/
catch(Exception e){ /* do nothing */ }
continue;
}
}
catch(Exception e)
{
// Ignore the failure raised while proceeding the root failure
}
}
// We are always done for this export ...
// ... except in the NewAttempt case (see above)
done = true;
} // attempt loop
// Wait for all servers to be ok, and build the failure list
// Servers that left in the list are those for which we could not test
// that they have been successfully initialized.
{
}
// Don't forget to release IEcontext acquired at beginning.
releaseIEContext(); // FIXME should not this be in a finally?
{
}
else
{
}
if (exportRootException != null)
{
throw exportRootException;
}
}
/**
* For all remote servers in the start list:
* - wait it has finished the import and present the expected generationID,
* - build the failureList.
*/
{
if (logger.isTraceEnabled())
"[IE] wait for start replicasWeAreWaitingFor=" + replicasWeAreWaitingFor);
int waitResultAttempt = 0;
boolean done;
do
{
done = true;
{
if (logger.isTraceEnabled())
+ " " + getGenerationID());
{
{
// this one is still not doing the Full Update ... retry later
done = false;
}
catch (InterruptedException e) {
}
break;
}
else
{
// this one is ok
}
}
}
}
if (logger.isTraceEnabled())
}
/**
* For all remote servers in the start list:
* - wait it has finished the import and present the expected generationID,
* - build the failureList.
*/
{
if (logger.isTraceEnabled())
"[IE] wait for end replicasWeAreWaitingFor=" + replicasWeAreWaitingFor);
/*
In case some new servers appear during the init, we want them to be
considered in the processing of sorting the successfully initialized
and the others
*/
boolean done;
do
{
done = true;
int reconnectMaxDelayInSec = 10;
int reconnectWait = 0;
{
{
/*
this server has already been in error during initialization
don't wait for it
*/
continue;
}
{
/*
this server is disconnected
may be for a long time if it crashed or had been stopped
may be just the time to reconnect after import : should be short
*/
{
// let's still wait to give a chance to this server to reconnect
done = false;
}
// Else we left enough time to the servers to reconnect
}
else
{
// this server is connected
{
// this one is still doing the Full Update ... retry later
done = false;
break;
}
{ // and with the expected generationId
// We're done with this server
}
}
}
// loop and wait
if (!done)
{
catch (InterruptedException e) {
} // 1sec
}
}
if (logger.isTraceEnabled())
}
/**
* Get the ServerState maintained by the Concrete class.
*
* @return the ServerState maintained by the Concrete class.
*/
public ServerState getServerState()
{
return state;
}
/**
*/
throws DirectoryException
{
{
// Rejects 2 simultaneous exports
}
return ieCtx;
}
private void releaseIEContext()
{
}
/**
* Processes an error message received while an export is
* on going, or an import will start.
*
* @param errorMsg The error message received.
*/
{
//Exporting must not be stopped on the first error, if we run initialize-all
{
// The ErrorMsg is received while we have started an initialization
/*
* This can happen :
* - on the first InitReqMsg sent when source in not known for example
* - on the next attempt when source crashed and did not reconnect
* even after the nextInitAttemptDelay
* During the import, the ErrorMsg will be received by receiveEntryBytes
*/
{
// Update the task that initiated the import
}
}
}
/**
* 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
*/
protected byte[] receiveEntryBytes()
{
while (true)
{
try
{
// In the context of the total update, we don't want any automatic
// re-connection done transparently by the broker because of a better
// RS or because of a connection failure.
// We want to be notified of topology change in order to track a
// potential disconnection of the exporter.
if (logger.isTraceEnabled())
{
+ ", receiveEntryBytes " + msg);
}
{
if (broker.shuttingDown())
{
// The server is in the shutdown process
return null;
}
else
{
// Handle connection issues
return null;
}
}
// Check good ordering of msg received
{
if (ieCtx.exporterProtocolVersion >=
{
// check the msgCnt of the msg received to check ordering
{
return null;
}
// send the ack of flow control mgmt
{
if (logger.isTraceEnabled())
{
+ ", publish InitializeRcvAckMsg" + amsg);
}
}
}
return entryBytes;
}
{
/*
This is the normal termination of the import
No error is stored and the import is ended by returning null
*/
return null;
}
{
/*
This is an error termination during the import
The error is stored and the import is ended by returning null
*/
{
{
return null;
}
}
}
else
{
// Other messages received during an import are trashed except
// the topologyMsg.
if (msg instanceof TopologyMsg
{
return null;
}
}
}
catch(Exception e)
{
}
}
}
/**
* Count the number of entries in the provided byte[].
* This is based on the hypothesis that the entries are separated
* by a "\n\n" String.
*
* @param entryBytes the set of bytes containing one or more entries.
* @return The number of entries in the provided byte[].
*/
private int countEntryLimits(byte[] entryBytes)
{
}
/**
* Count the number of entries in the provided byte[].
* This is based on the hypothesis that the entries are separated
* by a "\n\n" String.
*
* @param entryBytes the set of bytes containing one or more entries.
* @return The number of entries in the provided byte[].
*/
{
int entryCount = 0;
int count = 0;
{
{
entryCount++;
count++;
}
count++;
}
return entryCount;
}
/**
* Exports an entry in LDIF format.
*
* @param lDIFEntry The entry to be exported in byte[] form.
* @param pos The starting Position in the array.
* @param length Number of array elements to be copied.
*
* @throws IOException when an error occurred.
*/
throws IOException
{
if (logger.isTraceEnabled())
// build the message
// Waiting the slowest loop
while (!broker.shuttingDown())
{
/*
If an error was raised - like receiving an ErrorMsg from a remote
server that have been stored by the listener thread in the ieContext,
we just abandon the export by throwing an exception.
*/
{
}
{
throw new IOException("IOException with nested DirectoryException",
ieCtx.getException());
}
if (logger.isTraceEnabled())
{
if (logger.isTraceEnabled())
// our export is too far beyond the slowest importer - let's wait
catch(Exception e) { /* do nothing */ }
// process any connection error
if (broker.hasConnectionError()
{
// publish failed - store the error in the ieContext ...
// .. and abandon the export by throwing an exception.
}
}
else
{
if (logger.isTraceEnabled())
break;
}
} // Waiting the slowest loop
if (logger.isTraceEnabled())
// process any publish error
if (!sent
|| broker.hasConnectionError()
{
// publish failed - store the error in the ieContext ...
// .. and abandon the export by throwing an exception.
}
// publish succeeded
try
{
}
catch (DirectoryException de)
{
// .. and abandon the export by throwing an exception.
}
}
/**
* Initializes asynchronously this domain from a remote source server.
* Before returning from this call, for the provided task :
* - the progressing counters are updated during the initialization using
* setTotal() and setLeft().
* - the end of the initialization using updateTaskCompletionState().
* <p>
* When this method is called, a request for initialization is sent to the
* remote source server requesting initialization.
* <p>
*
* @param source The server-id of the source from which to initialize.
* The source can be discovered using the
* {@link #getReplicaInfos()} method.
*
* @param initTask The task that launched the initialization
* and should be updated of its progress.
*
* @throws DirectoryException If it was not possible to publish the
* Initialization message to the Topology.
* The task state is updated.
*/
throws DirectoryException
{
if (logger.isTraceEnabled())
{
}
: null;
/*
We must not test here whether the remote source is connected to
the topology by testing if it stands in the replicas list since.
In the case of a re-attempt of initialization, the listener thread is
running this method directly coming from initialize() method and did
not processed any topology message in between the failure and the
new attempt.
*/
try
{
/*
We must immediately acquire a context to store the task inside
The context will be used when we (the listener thread) will receive
the InitializeTargetMsg, process the import, and at the end
update the task.
*/
/*
The normal success processing is now to receive InitTargetMsg then
entries from the remote server.
The error cases are :
- either local error immediately caught below
- a remote error we will receive as an ErrorMsg
*/
}
catch(DirectoryException de)
{
}
catch(Exception e)
{
// Should not happen
}
// When error, update the task and raise the error to the caller
{
// No need to call here updateTaskCompletionState - will be done
// by the caller
}
}
/**
* Processes an InitializeTargetMsg received from a remote server
* meaning processes an initialization from the entries expected to be
* received now.
*
* @param initTargetMsgReceived The message received from the remote server.
*
* @param requesterServerId The serverId of the server that requested the
* initialization meaning the server where the
* task has initially been created (this server,
* or the remote server).
*/
{
if (logger.isTraceEnabled())
{
}
try
{
// Log starting
// Go into full update status
// Acquire an import context if no already done (and initialize).
{
/*
The initTargetMsgReceived is for an import initiated by the remote server.
Test and set if no import already in progress
*/
ieCtx = acquireIEContext(true);
}
// Initialize stuff
// Launch the import
importBackend(new ReplInputStream(this));
}
catch (DirectoryException e)
{
/*
Store the exception raised. It will be considered if no other exception
has been previously stored in the context
*/
}
finally
{
if (logger.isTraceEnabled())
{
}
/*
It is necessary to restart (reconnect to RS) for different reasons
- when everything went well, reconnect in order to exchange
new state, new generation ID
- when we have connection failure, reconnect to retry a new import
right here, right now
we never want retryOnFailure if we fails reconnecting in the restart.
*/
&& broker.isConnected()
&& initFromTask != null
{
/*
Worth a new attempt
since initFromTask is in this server, connection is ok
*/
try
{
/*
Wait for the exporter to stabilize - eventually reconnect as
well if it was connected to the same RS than the one we lost ...
*/
/*
Restart the whole import protocol exchange by sending again
the request
*/
// Processing of the received initTargetMsgReceived is done
// let's wait for the next one
return;
}
catch(Exception e)
{
/*
An error occurs when sending a new request for a new import.
This error is not stored, preferring to keep the initial one.
*/
}
}
// ===================
// No new attempt case
if (logger.isTraceEnabled())
{
+ " task=" + initFromTask
}
try
{
{
// Let's notify the exporter
}
/*
Update the task that initiated the import must be the last thing.
Particularly, broker.restart() after import success must be done
before some other operations/tasks to be launched,
like resetting the generation ID.
*/
if (initFromTask != null)
{
}
}
finally
{
} // finally
} // finally
}
/**
* Return the protocol version of the DS related to the provided serverId.
* Returns -1 when the protocol version is not known.
* @param dsServerId The provided serverId.
* @return The protocol version.
*/
private short getProtocolVersion(int dsServerId)
{
{
return dsInfo.getProtocolVersion();
}
return -1;
}
/**
* Sets the status to a new value depending of the passed status machine
* event.
* @param event The event that may make the status be changed
*/
{
}
{
{
return;
}
{
// Reset status date
lastStatusChangeDate = new Date();
// Reset monitoring counters if reconnection
{
}
if (logger.isTraceEnabled())
{
+ " new status is: " + status);
}
// Perform whatever actions are needed to apply properties for being
// compliant with new status
}
}
/**
* Returns a boolean indicating if an import or export is currently
* processed.
*
* @return The status
*/
public boolean ieRunning()
{
}
/**
* Check the value of the Replication Servers generation ID.
*
* @param generationID The expected value of the generation ID.
*
* @throws DirectoryException When the generation ID of the Replication
* Servers is not the expected value.
*/
{
boolean allSet = true;
for (int i = 0; i< 50; i++)
{
allSet = true;
{
// the 'empty' RSes (generationId==-1) are considered as good citizens
{
try
{
} catch (InterruptedException e)
{
}
allSet = false;
break;
}
}
if (allSet)
{
break;
}
}
if (!allSet)
{
}
}
/**
* Reset the Replication Log.
* Calling this method will remove all the Replication information that
* was kept on all the Replication Servers currently connected in the
* topology.
*
* @throws DirectoryException If this ReplicationDomain is not currently
* connected to a Replication Server or it
* was not possible to contact it.
*/
void resetReplicationLog() throws DirectoryException
{
// Reset the Generation ID to -1 to clean the ReplicationServers.
resetGenerationId(-1L);
// check that at least one ReplicationServer did change its generation-id
checkGenerationID(-1);
// Reconnect to the Replication Server so that it adopts our GenerationID.
// wait for the domain to reconnect.
int count = 0;
{
try
{
} catch (InterruptedException e)
{
}
}
// check that at least one ReplicationServer did change its generation-id
}
/**
* 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.
* @throws DirectoryException When an error occurs
*/
throws DirectoryException
{
if (logger.isTraceEnabled())
{
}
if (!isConnected())
{
}
// check that at least one ReplicationServer did change its generation-id
}
{
if (generationIdNewValue != null)
{
return generationIdNewValue;
}
return getGenerationID();
}
/*
******** End of The total Update code *********
*/
/*
******* Start of Monitoring Code **********
*/
/**
* Get the maximum receive window size.
*
* @return The maximum receive window size.
*/
int getMaxRcvWindow()
{
{
return broker.getMaxRcvWindow();
}
return 0;
}
/**
* Get the current receive window size.
*
* @return The current receive window size.
*/
int getCurrentRcvWindow()
{
{
return broker.getCurrentRcvWindow();
}
return 0;
}
/**
* Get the maximum send window size.
*
* @return The maximum send window size.
*/
int getMaxSendWindow()
{
{
return broker.getMaxSendWindow();
}
return 0;
}
/**
* Get the current send window size.
*
* @return The current send window size.
*/
int getCurrentSendWindow()
{
{
return broker.getCurrentSendWindow();
}
return 0;
}
/**
* Get the number of times the replication connection was lost.
* @return The number of times the replication connection was lost.
*/
int getNumLostConnections()
{
{
return broker.getNumLostConnections();
}
return 0;
}
/**
* Determine whether the connection to the replication server is encrypted.
* @return true if the connection is encrypted, false otherwise.
*/
boolean isSessionEncrypted()
{
}
/**
* Check if the domain is connected to a ReplicationServer.
*
* @return true if the server is connected, false if not.
*/
public boolean isConnected()
{
}
/**
* Check if the domain has a connection error.
* A Connection error happens when the broker could not be created
* or when the broker could not find any ReplicationServer to connect to.
*
* @return true if the domain has a connection error.
*/
public boolean hasConnectionError()
{
}
/**
* 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();
}
return ReplicationBroker.NO_CONNECTED_SERVER;
}
/**
* Gets the number of updates sent in assured safe read mode.
* @return The number of updates sent in assured safe read mode.
*/
public int getAssuredSrSentUpdates()
{
return assuredSrSentUpdates.get();
}
/**
* Gets the number of updates sent in assured safe read mode that have been
* acknowledged without errors.
* @return The number of updates sent in assured safe read mode that have been
* acknowledged without errors.
*/
public int getAssuredSrAcknowledgedUpdates()
{
return assuredSrAcknowledgedUpdates.get();
}
/**
* Gets the number of updates sent in assured safe read mode that have not
* been acknowledged.
* @return The number of updates sent in assured safe read mode that have not
* been acknowledged.
*/
public int getAssuredSrNotAcknowledgedUpdates()
{
return assuredSrNotAcknowledgedUpdates.get();
}
/**
* Gets the number of updates sent in assured safe read mode that have not
* been acknowledged due to timeout error.
* @return The number of updates sent in assured safe read mode that have not
* been acknowledged due to timeout error.
*/
public int getAssuredSrTimeoutUpdates()
{
return assuredSrTimeoutUpdates.get();
}
/**
* Gets the number of updates sent in assured safe read mode that have not
* been acknowledged due to wrong status error.
* @return The number of updates sent in assured safe read mode that have not
* been acknowledged due to wrong status error.
*/
public int getAssuredSrWrongStatusUpdates()
{
return assuredSrWrongStatusUpdates.get();
}
/**
* Gets the number of updates sent in assured safe read mode that have not
* been acknowledged due to replay error.
* @return The number of updates sent in assured safe read mode that have not
* been acknowledged due to replay error.
*/
public int getAssuredSrReplayErrorUpdates()
{
return assuredSrReplayErrorUpdates.get();
}
/**
* Gets the number of updates sent in assured safe read mode that have not
* been acknowledged per server.
* @return A copy of the map that contains the number of updates sent in
* assured safe read mode that have not been acknowledged per server.
*/
{
synchronized(assuredSrServerNotAcknowledgedUpdates)
{
}
}
/**
* Gets the number of updates received in assured safe read mode request.
* @return The number of updates received in assured safe read mode request.
*/
public int getAssuredSrReceivedUpdates()
{
return assuredSrReceivedUpdates.get();
}
/**
* Gets the number of updates received in assured safe read mode that we acked
* without error (no replay error).
* @return The number of updates received in assured safe read mode that we
* acked without error (no replay error).
*/
public int getAssuredSrReceivedUpdatesAcked()
{
return this.assuredSrReceivedUpdatesAcked.get();
}
/**
* Gets the number of updates received in assured safe read mode that we did
* not ack due to error (replay error).
* @return The number of updates received in assured safe read mode that we
* did not ack due to error (replay error).
*/
public int getAssuredSrReceivedUpdatesNotAcked()
{
return this.assuredSrReceivedUpdatesNotAcked.get();
}
/**
* Gets the number of updates sent in assured safe data mode.
* @return The number of updates sent in assured safe data mode.
*/
public int getAssuredSdSentUpdates()
{
return assuredSdSentUpdates.get();
}
/**
* Gets the number of updates sent in assured safe data mode that have been
* acknowledged without errors.
* @return The number of updates sent in assured safe data mode that have been
* acknowledged without errors.
*/
public int getAssuredSdAcknowledgedUpdates()
{
return assuredSdAcknowledgedUpdates.get();
}
/**
* Gets the number of updates sent in assured safe data mode that have not
* been acknowledged due to timeout error.
* @return The number of updates sent in assured safe data mode that have not
* been acknowledged due to timeout error.
*/
public int getAssuredSdTimeoutUpdates()
{
return assuredSdTimeoutUpdates.get();
}
/**
* Gets the number of updates sent in assured safe data mode that have not
* been acknowledged due to timeout error per server.
* @return A copy of the map that contains the number of updates sent in
* assured safe data mode that have not been acknowledged due to timeout
* error per server.
*/
{
synchronized(assuredSdServerTimeoutUpdates)
{
}
}
/**
* Gets the date of the last status change.
* @return The date of the last status change.
*/
public Date getLastStatusChangeDate()
{
return lastStatusChangeDate;
}
/**
* Resets the values of the monitoring counters.
*/
private void resetMonitoringCounters()
{
synchronized (assuredSrServerNotAcknowledgedUpdates)
{
}
synchronized (assuredSdServerTimeoutUpdates)
{
}
}
/*
********** End of Monitoring Code **************
*/
/**
* Start the publish mechanism of the Replication Service. After this method
* has been called, the publish service can be used by calling the
* {@link #publish(UpdateMsg)} method.
*
* @throws ConfigException
* If the DirectoryServer configuration was incorrect.
*/
public void startPublishService() throws ConfigException
{
synchronized (sessionLock)
{
{
// create the broker object used to publish and receive changes
broker = new ReplicationBroker(
}
}
}
/**
* Starts the receiver side of the Replication Service.
* <p>
* After this method has been called, the Replication Service will start
* calling the {@link #processUpdate(UpdateMsg)}.
* <p>
* This method must be called once and must be called after the
* {@link #startPublishService()}.
*/
public void startListenService()
{
synchronized (sessionLock)
{
{
public void run()
{
if (logger.isTraceEnabled())
{
}
// Loop processing any incoming update messages.
while (!listenerThread.isShutdownInitiated())
{
{
// The server is shutting down.
}
else if (processUpdate(updateMsg)
{
/*
* Warning: in synchronous mode, no way to tell the replay of an
* update went wrong Just put null in processUpdateDone so that if
* assured replication is used the ack is sent without error at
* replay flag.
*/
}
}
if (logger.isTraceEnabled())
{
}
}
}, threadName);
}
}
/**
* Temporarily disable the Replication Service.
* The Replication Service can be enabled again using
* {@link #enableService()}.
* <p>
* It can be useful to disable the Replication Service when the
* repository where the replicated information is stored becomes
* temporarily unavailable and replicated updates can therefore not
* be replayed during a while. This method is not MT safe.
*/
public void disableService()
{
synchronized (sessionLock)
{
/*
Stop the broker first in order to prevent the listener from
reconnecting - see OPENDJ-457.
*/
{
}
// Stop the listener thread
if (listenerThread != null)
{
try
{
}
catch (InterruptedException e)
{
// Give up waiting.
}
}
}
}
/**
* Returns {@code true} if the listener thread is shutting down or has
* shutdown.
*
* @return {@code true} if the listener thread is shutting down or has
* shutdown.
*/
protected final boolean isListenerShuttingDown()
{
}
/**
* Restart the Replication service after a {@link #disableService()}.
* <p>
* The Replication Service will restart from the point indicated by the
* {@link ServerState} that was given as a parameter to the
* {@link #startPublishService()} at startup time.
* <p>
* If some data have changed in the repository during the period of time when
* the Replication Service was disabled, this {@link ServerState} should
* therefore be updated by the Replication Domain subclass before calling this
* method. This method is not MT safe.
*/
public void enableService()
{
synchronized (sessionLock)
{
}
}
/**
* Change some ReplicationDomain parameters.
*
* @param config
* The new configuration that this domain should now use.
*/
{
{
}
}
/**
* Applies a configuration change to the attributes which should be included
* in the ECL.
*
* @param includeAttributes
* attributes to be included with all change records.
* @param includeAttributesForDeletes
* additional attributes to be included with delete change records.
*/
{
final boolean attrsModified = setEclIncludes(
{
}
}
private void restartService()
{
}
/**
* This method should trigger an export of the replicated data.
* to the provided outputStream.
* When finished the outputStream should be flushed and closed.
*
* @param output The OutputStream where the export should
* be produced.
* @throws DirectoryException When needed.
*/
throws DirectoryException;
/**
* This method should trigger an import of the replicated data.
*
* @param input The InputStream from which
* the import should be reading entries.
*
* @throws DirectoryException When needed.
*/
throws DirectoryException;
/**
* 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 abstract long countEntries() throws DirectoryException;
/**
* This method should handle the processing of {@link UpdateMsg} receive from
* remote replication entities.
* <p>
* This method will be called by a single thread and should therefore should
* not be blocking.
*
* @param updateMsg
* The {@link UpdateMsg} that was received.
* @return A boolean indicating if the processing is completed at return time.
* If <code> true </code> is returned, no further processing is
* necessary. If <code> false </code> is returned, the subclass should
* call the method {@link #processUpdateDone(UpdateMsg, String)} and
* update the ServerState When this processing is complete.
*/
/**
* This method must be called after each call to
* {@link #processUpdate(UpdateMsg)} when the processing of the
* update is completed.
* <p>
* It is useful for implementation needing to process the update in an
* asynchronous way or using several threads, but must be called even by
* implementation doing it in a synchronous, single-threaded way.
*
* @param msg
* The UpdateMsg whose processing was completed.
* @param replayErrorMsg
* if not null, this means an error occurred during the replay of
* this update, and this is the matching human readable message
* describing the problem.
*/
{
/*
Send an ack if it was requested and the group id is the same of the RS
one. Only Safe Read mode makes sense in DS for returning an ack.
*/
// Assured feature is supported starting from replication protocol V2
{
{
{
// Send the ack
if (replayErrorMsg != null)
{
// Mark the error in the ack
// -> replay error occurred
ackMsg.setHasReplayError(true);
// -> replay error occurred in our server
}
if (replayErrorMsg != null)
{
}
else
{
}
}
}
{
}
// Nothing to do in Assured safe data mode, only RS ack updates.
}
}
/**
* Prepare a message if it is to be sent in assured mode.
* If the assured mode is enabled, this method should be called before
* publish(UpdateMsg msg) method. This will configure the update accordingly
* before it is sent and will prepare the mechanism that will block until the
* matching ack is received. To wait for the ack after publish call, use
* the waitForAckIfAssuredEnabled() method.
* The expected typical usage in a service inheriting from this class is
* the following sequence:
* UpdateMsg msg = xxx;
* prepareWaitForAckIfAssuredEnabled(msg);
* publish(msg);
* waitForAckIfAssuredEnabled(msg);
*
* Note: prepareWaitForAckIfAssuredEnabled and waitForAckIfAssuredEnabled have
* no effect if assured replication is disabled.
* Note: this mechanism should not be used if using publish(byte[] msg)
* version as usage of these methods is already hidden inside.
*
* @param msg The update message to be sent soon.
*/
{
/*
* If assured configured, set message accordingly to request an ack in the
* right assured mode.
* No ack requested for a RS with a different group id.
* Assured replication supported for the same locality,
* i.e: a topology working in the same geographical location).
* If we are connected to a RS which is not in our locality,
* no need to ask for an ack.
*/
if (needsAck())
{
msg.setAssured(true);
{
}
// Add the assured message to the list of update that are waiting for acks
}
}
private boolean needsAck()
{
}
/**
* Wait for the processing of an assured message after it has been sent, if
* assured replication is configured, otherwise, do nothing.
* The prepareWaitForAckIfAssuredEnabled method should have been called
* before, see its comment for the full picture.
*
* @param msg The UpdateMsg for which we are waiting for an ack.
* @throws TimeoutException When the configured timeout occurs waiting for the
* ack.
*/
throws TimeoutException
{
if (needsAck())
{
// Increment assured replication monitoring counters
switch (getAssuredMode())
{
case SAFE_READ_MODE:
break;
case SAFE_DATA_MODE:
break;
default:
// Should not happen
}
} else
{
// Not assured or bad group id, return immediately
return;
}
// Wait for the ack to be received, timing out if necessary
synchronized (msg)
{
{
try
{
/*
WARNING: this timeout may be difficult to optimize: too low, it
may use too much CPU, too high, it may penalize performance...
*/
} catch (InterruptedException e)
{
if (logger.isTraceEnabled())
{
"baseDN: " + getBaseDN());
}
break;
}
// Timeout ?
{
/*
Timeout occurred, be sure that ack is not being received and if so,
remove the update from the wait list, log the timeout error and
also update assured monitoring counters
*/
{
// Ack received just before timeout limit: we can exit
break;
}
// No luck, this is a real timeout
// Increment assured replication monitoring counters
switch (msg.getAssuredMode())
{
case SAFE_READ_MODE:
// Increment number of errors for our RS
broker.getRsServerId());
break;
case SAFE_DATA_MODE:
// Increment number of errors for our RS
broker.getRsServerId());
break;
default:
// Should not happen
}
+ getAssuredTimeout() + " ms.");
}
}
}
}
/**
* Publish an {@link UpdateMsg} to the Replication Service.
* <p>
* The Replication Service will handle the delivery of this {@link UpdateMsg}
* to all the participants of this Replication Domain. These members will be
* receive this {@link UpdateMsg} through a call of the
* {@link #processUpdate(UpdateMsg)} message.
*
* @param msg The UpdateMsg that should be published.
*/
{
if (msg.contributesToDomainState())
{
}
}
/**
* Publishes a replica offline message if all pending changes for current
* replica have been sent out.
*/
public void publishReplicaOfflineMsg()
{
// Here to be overridden
}
/**
* This method should return the generationID to use for this
* ReplicationDomain.
* This method can be called at any time after the ReplicationDomain
* has been started.
*
* @return The GenerationID.
*/
public long getGenerationID()
{
return generationId;
}
/**
* Sets the generationId for this replication domain.
*
* @param generationId
* the generationId to set
*/
public void setGenerationID(long generationId)
{
this.generationId = generationId;
}
/**
* Subclasses should use this method to add additional monitoring information
* in the ReplicationDomain.
*
* @return Additional monitoring attributes that will be added in the
* ReplicationDomain monitoring entry.
*/
{
}
/**
*
*/
protected ImportExportContext getImportExportContext()
{
return importExportContext.get();
}
/**
* Returns the local address of this replication domain, or the empty string
* if it is not yet connected.
*
* @return The local address.
*/
{
}
/**
* Set the attributes configured on a server to be included in the ECL.
*
* @param serverId
* Server where these attributes are configured.
* @param includeAttributes
* Attributes to be included with all change records, may include
* wild-cards.
* @param includeAttributesForDeletes
* Additional attributes to be included with delete change records,
* may include wild-cards.
* @return {@code true} if the set of attributes was modified.
*/
public boolean setEclIncludes(int serverId,
{
do
{
}
}
/**
* Get the attributes to include in each change for the ECL.
*
* @return The attributes to include in each change for the ECL.
*/
{
}
/**
* Get the attributes to include in each delete change for the ECL.
*
* @return The attributes to include in each delete change for the ECL.
*/
{
}
/**
* Get the attributes to include in each change for the ECL for a given
* serverId.
*
* @param serverId
* The serverId for which we want the include attributes.
* @return The attributes.
*/
{
}
/**
* Get the attributes to include in each change for the ECL for a given
* serverId.
*
* @param serverId
* The serverId for which we want the include attributes.
* @return The attributes.
*/
{
}
/**
* Returns the CSN of the last Change that was fully processed by this
* ReplicationDomain.
*
* @return The CSN of the last Change that was fully processed by this
* ReplicationDomain.
*/
public CSN getLastLocalChange()
{
}
/**
* Gets and stores the assured replication configuration parameters. Returns a
* boolean indicating if the passed configuration has changed compared to
* previous values and the changes require a reconnection.
*
* @param config
* The configuration object
* @param allowReconnection
* Tells if one must reconnect if significant changes occurred
*/
boolean allowReconnection)
{
// Disconnect if required: changing configuration values before
// disconnection would make assured replication used immediately and
// disconnection could cause some timeouts error.
{
}
}
{
switch (cfg.getAssuredType())
{
case NOT_ASSURED:
if (isAssured())
{
return true;
}
break;
case SAFE_DATA:
{
return true;
}
break;
case SAFE_READ:
{
return true;
}
break;
}
return isAssured()
&& assuredMode == SAFE_DATA_MODE
}
/** {@inheritDoc} */
{
}
}