com.unboundid.directory.sdk.sync.api

Class ChangeDetector

java.lang.Object

com.unboundid.directory.sdk.sync.api.ChangeDetector

All Implemented Interfaces:

Configurable, ExampleUsageProvider, Reconfigurable<ChangeDetectorConfig>, UnboundIDExtension

@Extensible
 @SynchronizationServerExtension(appliesToLocalContent=false,
                                appliesToSynchronizedContent=true)
public abstract class ChangeDetector
extends java.lang.Object
implements UnboundIDExtension, Reconfigurable<ChangeDetectorConfig>, ExampleUsageProvider

This class defines an API that must be implemented by extensions that detect changes for an LDAP based Sync Source. A Change Detector can be used to

• Process logs or other flat files for changes.
• Process changes from a Queue (Kafka, RabbitMQ, etc)
• Override the standard cn=changelog based approach for detecting changes.

Configuring Change Detectors

In order to configure a Change Detector created using this API, use a command like:

    dsconfig create-change-detector \
          --detector-name "{detector-name}" \
          --type third-party \
          --set "extension-class:{class-name}" \
          --set "extension-argument:{name=vale}"
 

where "{plugin-name}" is the name to use for the Change Detector instance, "{class-name}" is the fully-qualified name of the Java class that extends com.unboundid.directory.sdk.sync.api.ChangeDetector, and "{name=value}" represents name-value pairs for any arguments to provide to the Change Detector. If multiple arguments should be provided to the Change Detector, then the "--set extension-argument:{name=value}" option should be provided multiple times.

Constructor Summary

Constructors

Constructor and Description
ChangeDetector() Creates a new instance of this LDAP Change Detector.

Method Summary

Modifier and Type	Method and Description
abstract void	acknowledgeCompletedOps(java.util.LinkedList<SyncOperation> completedOps) Provides a way for the Data Sync Server to acknowledge back to the extension which sync operations it has processed.
ResultCode	applyConfiguration(ChangeDetectorConfig config, ArgumentParser parser, java.util.List<java.lang.String> adminActionsRequired, java.util.List<java.lang.String> messages) Attempts to apply the configuration from the provided argument parser to this extension.
void	defineConfigArguments(ArgumentParser parser) Updates the provided argument parser to define any configuration arguments which may be used by this extension.
void	finalizeChangeDetector() This hook is called when a Sync Pipe shuts down or when the set-startpoint subcommand (from the realtime-sync command line tool) is finished.
java.util.Map<java.util.List<java.lang.String>,java.lang.String>	getExamplesArgumentSets() Retrieves a map containing examples of configurations that may be used for this extension.
abstract java.lang.String[]	getExtensionDescription() Retrieves a human-readable description for this extension.
abstract java.lang.String	getExtensionName() Retrieves a human-readable name for this extension.
abstract java.util.List<ChangeRecord>	getNextBatchOfChanges(int maxChanges, java.util.concurrent.atomic.AtomicLong numStillPending) Return the next batch of change records from the source.
abstract java.io.Serializable	getStartpoint() Gets the current value of the startpoint for change detection.
void	initializeChangeDetector(SyncServerContext serverContext, SyncSourceContext syncSourceContext, ArgumentParser parser) This hook is called when a Sync Pipe first starts up or when the set-startpoint subcommand is called from the realtime-sync command line tool.
boolean	isConfigurationAcceptable(ChangeDetectorConfig config, ArgumentParser parser, java.util.List<java.lang.String> unacceptableReasons) Indicates whether the configuration represented by the provided argument parser is acceptable for use by this extension.
abstract void	setStartpoint(SetStartpointOptions options) This method should effectively set the starting point for synchronization to the place specified by the options parameter.

Methods inherited from class java.lang.Object

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

Constructor Detail

ChangeDetector

public ChangeDetector()

Creates a new instance of this LDAP Change Detector. All Change Detector implementations must include a default constructor, but any initialization should generally be done in the initializeChangeDetector method.

Method Detail

getExtensionName

public abstract java.lang.String getExtensionName()

Retrieves a human-readable name for this extension.

Specified by:

getExtensionName in interface UnboundIDExtension

Returns:

A human-readable name for this extension.

getExtensionDescription

public abstract java.lang.String[] getExtensionDescription()

Retrieves a human-readable description for this extension. Each element of the array that is returned will be considered a separate paragraph in generated documentation.

Specified by:

getExtensionDescription in interface UnboundIDExtension

Returns:

A human-readable description for this extension, or null or an empty array if no description should be available.

getExamplesArgumentSets

public java.util.Map<java.util.List<java.lang.String>,java.lang.String> getExamplesArgumentSets()

Retrieves a map containing examples of configurations that may be used for this extension. The map key should be a list of sample arguments, and the corresponding value should be a description of the behavior that will be exhibited by the extension when used with that configuration.

Specified by:

getExamplesArgumentSets in interface ExampleUsageProvider

Returns:

A map containing examples of configurations that may be used for this extension. It may be null or empty if there should not be any example argument sets.

defineConfigArguments

public void defineConfigArguments(ArgumentParser parser)
                           throws ArgumentException

Updates the provided argument parser to define any configuration arguments which may be used by this extension. The argument parser may also be updated to define relationships between arguments (e.g., to specify required, exclusive, or dependent argument sets).

Specified by:

defineConfigArguments in interface Configurable

Parameters:

parser - The argument parser to be updated with the configuration arguments which may be used by this extension.

Throws:

ArgumentException - If a problem is encountered while updating the provided argument parser.

initializeChangeDetector

public void initializeChangeDetector(SyncServerContext serverContext,
                                     SyncSourceContext syncSourceContext,
                                     ArgumentParser parser)

This hook is called when a Sync Pipe first starts up or when the set-startpoint subcommand is called from the realtime-sync command line tool. Any initialization of this change detector should be performed here. This method should generally store the SyncServerContext and SyncSourceContext in a class member so that it can be used elsewhere in the implementation.

The default implementation is empty.

Parameters:

serverContext - A handle to the server context for the server in which this extension is running.

syncSourceContext - An interface for interacting with the Sync Source that owns this Change Detector or null if the Change Detector is only being initialized to validate its configuration.

parser - The argument parser which has been initialized from the configuration for this sync source.

finalizeChangeDetector

public void finalizeChangeDetector()

This hook is called when a Sync Pipe shuts down or when the set-startpoint subcommand (from the realtime-sync command line tool) is finished. Any clean up of this change detector should be performed here.

The default implementation is empty.

isConfigurationAcceptable

public boolean isConfigurationAcceptable(ChangeDetectorConfig config,
                                         ArgumentParser parser,
                                         java.util.List<java.lang.String> unacceptableReasons)

Indicates whether the configuration represented by the provided argument parser is acceptable for use by this extension. The parser will have been used to parse any configuration available for this extension, and any automatic validation will have been performed. This method may be used to perform any more complex validation which cannot be performed automatically by the argument parser.

Specified by:

isConfigurationAcceptable in interface Reconfigurable<ChangeDetectorConfig>

Parameters:

config - The general configuration for this extension.

parser - The argument parser that has been used to parse the proposed configuration for this extension.

unacceptableReasons - A list to which messages may be added to provide additional information about why the provided configuration is not acceptable.

Returns:

true if the configuration in the provided argument parser appears to be acceptable, or false if not.

applyConfiguration

public ResultCode applyConfiguration(ChangeDetectorConfig config,
                                     ArgumentParser parser,
                                     java.util.List<java.lang.String> adminActionsRequired,
                                     java.util.List<java.lang.String> messages)

Attempts to apply the configuration from the provided argument parser to this extension.

Specified by:

applyConfiguration in interface Reconfigurable<ChangeDetectorConfig>

Parameters:

config - The general configuration for this extension.

parser - The argument parser that has been used to parse the new configuration for this extension.

adminActionsRequired - A list to which messages may be added to provide additional information about any additional administrative actions that may be required to apply some of the configuration changes.

messages - A list to which messages may be added to provide additional information about the processing performed by this method.

Returns:

A result code providing information about the result of applying the configuration change. A result of SUCCESS should be used to indicate that all processing completed successfully. Any other result will indicate that a problem occurred during processing.

setStartpoint

public abstract void setStartpoint(SetStartpointOptions options)
                            throws EndpointException

This method should effectively set the starting point for synchronization to the place specified by the options parameter. This should cause all changes previous to the specified start point to be disregarded and only changes after that point to be returned by getNextBatchOfChanges(int, AtomicLong).

There are several different startpoint types (see SetStartpointOptions), and this implementation is not required to support them all. If the specified startpoint type is unsupported, this method should throw an UnsupportedOperationException. IMPORTANT: The RESUME_AT_SERIALIZABLE startpoint type must be supported by your implementation, because this is used when a Sync Pipe first starts up. The Serializable in this case is the same type that is returned by getStartpoint(); the Sync Server persists it and passes it back in on a restart.

This method can be called from two different contexts:

• When the 'set-startpoint' subcommand of the realtime-sync CLI is used (the Sync Pipe is required to be stopped in this context)
• Immediately after a Sync Pipe starts up and a connection is first established to the source server (e.g. before the first call to getNextBatchOfChanges(int, AtomicLong))

Parameters:

options - an object which indicates where exactly to start synchronizing (e.g. the end of the changelog, specific change number, a certain time ago, etc)

Throws:

EndpointException - if there is any error while setting the start point

getStartpoint

public abstract java.io.Serializable getStartpoint()

Gets the current value of the startpoint for change detection. This is the "bookmark" which indicates which changes have already been processed and which have not. In most cases, a change number is used to detect changes and is managed by the Data Sync Server, in which case this implementation needs only to return the latest acknowledged change number. In other cases, the return value may correspond to a different value, such as the SYS_CHANGE_VERSION in Microsoft SQL Server. In any case, this method should return the value that is updated by acknowledgeCompletedOps(LinkedList).

This method is called periodically and the return value is saved in the persistent state for the Sync Pipe that uses this extension as its Sync Source. IMPORTANT: The internal value for the startpoint should only be updated after a sync operation is acknowledged back to this extension (via acknowledgeCompletedOps(LinkedList)). Otherwise it will be possible for changes to be missed when the Data Sync Server is restarted or a connection error occurs.

Returns:

a value to store in the persistent state for the Sync Pipe. This is usually a change number, but if a changelog table is not used to detect changes, this value should represent some other token to pass into setStartpoint(SetStartpointOptions) when the sync pipe starts up.

getNextBatchOfChanges

public abstract java.util.List<ChangeRecord> getNextBatchOfChanges(int maxChanges,
                                                                   java.util.concurrent.atomic.AtomicLong numStillPending)
                                                            throws EndpointException

Return the next batch of change records from the source. Change records are usually just hints that a change happened; they do not include the full contents of the target entry. In an effort to never synchronize stale data, the Data Sync Server will go back and fetch the full target entry for each change record.

On the first invocation, this should return changes starting from the startpoint that was set by setStartpoint(SetStartpointOptions). This method is also responsible for updating the internal state such that subsequent invocations do not return duplicate changes.

The resulting list should be limited by maxChanges. The numStillPending reference should be set to the estimated number of changes that haven't yet been retrieved from the source endpoint when this method returns, or zero if all the current changes have been retrieved. IMPORTANT: While this method needs to keep track of which changes have already been returned so that it does not return them again, it should NOT modify the official startpoint. The internal value for the startpoint should only be updated after a sync operation is acknowledged back to this extension (via acknowledgeCompletedOps(LinkedList)). Otherwise it will be possible for changes to be missed when the Data Sync Server is restarted or a connection error occurs. The startpoint should not change as a result of this method.

This method does not need to be thread-safe. It will be invoked repeatedly by a single thread, based on the polling interval set in the Sync Pipe configuration.

Parameters:

maxChanges - the maximum number of changes to retrieve

numStillPending - this should be set to the number of unretrieved changes that are still pending after this batch has been retrieved. This will be passed in as zero, and may be left that way if the actual value cannot be determined.

Returns:

a list of ChangeRecord instances, each corresponding to a single change at the source endpoint. If there are no new changes to return, this method should return an empty list.

Throws:

EndpointException - if there is any error while retrieving the next batch of changes

acknowledgeCompletedOps

public abstract void acknowledgeCompletedOps(java.util.LinkedList<SyncOperation> completedOps)
                                      throws EndpointException

Provides a way for the Data Sync Server to acknowledge back to the extension which sync operations it has processed. This method should update the official startpoint which was set by setStartpoint(SetStartpointOptions) and is returned by getStartpoint(). IMPORTANT: The internal value for the startpoint should only be updated after a sync operation is acknowledged back to this extension (via this method). Otherwise it will be possible for changes to be missed when the Data Sync Server is restarted or a connection error occurs.

Parameters:

completedOps - a list of SyncOperations that have finished processing. The records are listed in the order they were first detected.

Throws:

EndpointException - if there is an error acknowledging the changes back to the source