@Extensible @SynchronizationServerExtension(appliesToLocalContent=false, appliesToSynchronizedContent=true) public abstract class JDBCSyncSource extends java.lang.Object implements UnboundIDExtension, Configurable, ExampleUsageProvider
fetchEntry(TransactionContext, SyncOperation)
method will be
called once for every change that is returned by
getNextBatchOfChanges(TransactionContext, int, AtomicLong)
.
During realtime synchronization (i.e. when a Sync Pipe is running), there is a sliding window of changes being processed, and this API provides a distinction between some different points along that window:
getNextBatchOfChanges()
but not completely processed and
acknowledged back to the Sync Source.getNextBatchOfChanges()
should return the first changes
that have not been detected. This should be somewhere at or ahead of
the startpoint.
In several places a TransactionContext
is provided, which allows
controlled access to the target database. By default, methods in this class
are always provided with a fresh connection (i.e. a new transaction), and the
Data Sync Server will always commit or rollback the transaction
automatically, depending on how the method returns. If a method call returns
successfully, then the transaction will be committed. If an exception is
thrown, then the transaction will be rolled back. In rare situations, it
might be necessary for an implementation to perform its own commit or
rollback of transactions by calling methods on TransactionContext
.
Several of these methods throw SQLException
, which should be used in
the case of any database access error. For other types of errors, runtime
exceptions may be used (IllegalStateException, NullPointerException, etc.).
The Data Sync Server will automatically retry operations that fail,
up to a configurable amount of attempts. The exception to this rule is if a
SQLException is thrown with a SQL state string beginning with "08"; this
indicates a connection error, and in this case the operation is retried
indefinitely. For these reasons implementers should refrain from handling
or wrapping any SQLException
and instead let it be handled by the
calling code.
dsconfig create-sync-source \ --source-name "{source-name}" \ --type third-party-jdbc \ --set "server:{server-name}" \ --set "extension-class:{class-name}" \ --set "extension-argument:{name=value}"where "{source-name}" is the name to use for the JDBC sync source instance, "{server-name}" is the name of the JDBC external server that will be used as the sync source, "{class-name}" is the fully-qualified name of the Java class written using this API, and "{name=value}" represents name-value pairs for any arguments to provide to the JDBC sync source. If multiple arguments should be provided to the JDBC sync source, then the "
--set extension-argument:{name=value}
" option
should be provided multiple times.Constructor and Description |
---|
JDBCSyncSource()
Creates a new instance of this JDBC Sync Source.
|
Modifier and Type | Method and Description |
---|---|
abstract void |
acknowledgeCompletedOps(TransactionContext ctx,
java.util.LinkedList<SyncOperation> completedOps)
Provides a way for the Data Sync Server to acknowledge back to the
script which sync operations it has processed.
|
int |
cleanupChangelog(TransactionContext ctx,
long maxAgeMillis)
Performs a cleanup of the changelog table (if desired).
|
void |
defineConfigArguments(com.unboundid.util.args.ArgumentParser parser)
Updates the provided argument parser to define any configuration arguments
which may be used by this extension.
|
abstract com.unboundid.ldap.sdk.Entry |
fetchEntry(TransactionContext ctx,
SyncOperation operation)
Return a full source entry (in LDAP form) from the database, corresponding
to the
DatabaseChangeRecord that is passed in through the
SyncOperation . |
void |
finalizeJDBCSyncSource(TransactionContext ctx)
This hook is called when a Sync Pipe shuts down, when the resync
process 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<DatabaseChangeRecord> |
getNextBatchOfChanges(TransactionContext ctx,
int maxChanges,
java.util.concurrent.atomic.AtomicLong numStillPending)
Return the next batch of change records from the database.
|
abstract java.io.Serializable |
getStartpoint()
Gets the current value of the startpoint for change detection.
|
void |
initializeJDBCSyncSource(TransactionContext ctx,
SyncServerContext serverContext,
JDBCSyncSourceConfig config,
com.unboundid.util.args.ArgumentParser parser)
This hook is called when a Sync Pipe first starts up, when the
resync process first starts up, or when the set-startpoint
subcommand is called from the realtime-sync command line tool.
|
void |
listAllEntries(TransactionContext ctx,
java.util.Iterator<java.lang.String> inputLines,
java.util.concurrent.BlockingQueue<DatabaseChangeRecord> outputQueue)
Deprecated.
|
void |
listAllEntries(TransactionContext ctx,
java.lang.String entryType,
java.util.concurrent.BlockingQueue<DatabaseChangeRecord> outputQueue)
Gets a list of all the entries in the database for a given entry type.
|
void |
listAllEntries(TransactionContext ctx,
java.lang.String entryType,
java.util.Iterator<java.lang.String> inputLines,
java.util.concurrent.BlockingQueue<DatabaseChangeRecord> outputQueue)
Gets a list of all the entries in the database from a given file input.
|
abstract void |
setStartpoint(TransactionContext ctx,
SetStartpointOptions options)
This method should effectively set the starting point for synchronization
to the place specified by the
options parameter. |
public JDBCSyncSource()
initializeJDBCSyncSource(com.unboundid.directory.sdk.sync.types.TransactionContext, com.unboundid.directory.sdk.sync.types.SyncServerContext, com.unboundid.directory.sdk.sync.config.JDBCSyncSourceConfig, com.unboundid.util.args.ArgumentParser)
method.public abstract java.lang.String getExtensionName()
getExtensionName
in interface UnboundIDExtension
public abstract java.lang.String[] getExtensionDescription()
getExtensionDescription
in interface UnboundIDExtension
null
or an empty array if no description should be available.public java.util.Map<java.util.List<java.lang.String>,java.lang.String> getExamplesArgumentSets()
getExamplesArgumentSets
in interface ExampleUsageProvider
null
or empty if there should
not be any example argument sets.public void defineConfigArguments(com.unboundid.util.args.ArgumentParser parser) throws com.unboundid.util.args.ArgumentException
defineConfigArguments
in interface Configurable
parser
- The argument parser to be updated with the configuration
arguments which may be used by this extension.com.unboundid.util.args.ArgumentException
- If a problem is encountered while updating the
provided argument parser.@ThreadSafety(level=METHOD_NOT_THREADSAFE) public void initializeJDBCSyncSource(TransactionContext ctx, SyncServerContext serverContext, JDBCSyncSourceConfig config, com.unboundid.util.args.ArgumentParser parser)
SyncServerContext
in a class
member so that it can be used elsewhere in the implementation.
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
The default implementation is empty.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.serverContext
- A handle to the server context for the server in
which this extension is running.config
- The general configuration for this sync source.parser
- The argument parser which has been initialized from
the configuration for this JDBC sync source.@ThreadSafety(level=METHOD_NOT_THREADSAFE) public void finalizeJDBCSyncSource(TransactionContext ctx)
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
The default implementation is empty.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.@ThreadSafety(level=METHOD_NOT_THREADSAFE) public abstract void setStartpoint(TransactionContext ctx, SetStartpointOptions options) throws java.sql.SQLException
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(TransactionContext, 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.
This method can be called from two different contexts:
getNextBatchOfChanges(TransactionContext, int, AtomicLong)
)
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.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)java.sql.SQLException
- if there is any error while setting the start point@ThreadSafety(level=METHOD_NOT_THREADSAFE) public abstract java.io.Serializable getStartpoint()
acknowledgeCompletedOps(TransactionContext, LinkedList)
.
This method is called periodically and the return value is saved in the persistent state for the Sync Pipe that uses this script as its Sync Source.
IMPORTANT: The internal value for the startpoint should only be
updated after a sync operation is acknowledged back to this script (via
acknowledgeCompletedOps(TransactionContext, LinkedList)
).
Otherwise it will be possible for changes to be missed when the
Data Sync Server is restarted or a connection error occurs.
setStartpoint(TransactionContext, SetStartpointOptions)
when the sync pipe starts up.@ThreadSafety(level=METHOD_NOT_THREADSAFE) public abstract java.util.List<DatabaseChangeRecord> getNextBatchOfChanges(TransactionContext ctx, int maxChanges, java.util.concurrent.atomic.AtomicLong numStillPending) throws java.sql.SQLException
On the first invocation, this should return changes starting from the
startpoint that was set by
setStartpoint(TransactionContext, SetStartpointOptions)
. This
method is 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 changelog table
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 script (via
acknowledgeCompletedOps(TransactionContext, 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.
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
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.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.maxChanges
- the maximum number of changes to retrievenumStillPending
- 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.DatabaseChangeRecord
instances, each
corresponding
to a row in the changelog table (or the equivalent if some other
change
tracking mechanism is being used). If there are no new changes to
return, this
method should return an empty list.java.sql.SQLException
- if there is any error while retrieving the next batch of changes@ThreadSafety(level=METHOD_THREADSAFE) public abstract com.unboundid.ldap.sdk.Entry fetchEntry(TransactionContext ctx, SyncOperation operation) throws java.sql.SQLException
DatabaseChangeRecord
that is passed in through the
SyncOperation
. This method should perform any queries necessary to
gather the latest values for all the attributes to be synchronized.
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
This method must be thread safe, as it will be called repeatedly and concurrently by each of the Sync Pipe worker threads as they process entries.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.operation
- the SyncOperation which identifies the database "entry" to
fetch. The DatabaseChangeRecord can be obtained by calling
operation.getDatabaseChangeRecord()
.
This is returned by
getNextBatchOfChanges(TransactionContext, int, AtomicLong)
or by
listAllEntries(TransactionContext, String, BlockingQueue)
.java.sql.SQLException
- if there is an error fetching the entry@ThreadSafety(level=METHOD_NOT_THREADSAFE) public abstract void acknowledgeCompletedOps(TransactionContext ctx, java.util.LinkedList<SyncOperation> completedOps) throws java.sql.SQLException
setStartpoint(TransactionContext, 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 script (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.
A TransactionContext
is provided in case the acknowledgment needs
to make it all the way back to the database itself (for example if you were
using Oracle's Change Data Capture). The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.completedOps
- a list of SyncOperation
s that have finished processing.
The records are listed in the order they were first detected.java.sql.SQLException
- if there is an error acknowledging the changes back to the
database@ThreadSafety(level=METHOD_NOT_THREADSAFE) public int cleanupChangelog(TransactionContext ctx, long maxAgeMillis) throws java.sql.SQLException
maxAgeMillis
milliseconds old.
NOTE: If the system clock on the database server is not in sync with the system clock on the Data Sync Server, this method should query the database for its current time in order to determine the cut-off point for deleting changelog records.
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
If a separate mechanism will be used to manage the changelog table, this method may be implemented as a no-op and always return zero. This is how the default implementation behaves.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.maxAgeMillis
- the period of time (in milliseconds) after which a changelog table
record should be deletedjava.sql.SQLException
- if there is an error purging records from the changelog table@ThreadSafety(level=METHOD_NOT_THREADSAFE) public void listAllEntries(TransactionContext ctx, java.lang.String entryType, java.util.concurrent.BlockingQueue<DatabaseChangeRecord> outputQueue) throws java.sql.SQLException
UnsupportedOperationException
; subclasses should override
if the resync functionality is needed.
The entryType
is user-defined; it will be
passed in on the command line for resync. The outputQueue
should contain DatabaseChangeRecord
objects with the
ChangeType
set to resync.
This method should not return until all the entries of the given entryType
have been added to the output queue. Separate threads will concurrently
drain entries from the queue and process them. The queue should not
actually contain full entries, but rather DatabaseChangeRecord objects
which identify the full database entries. These objects are then
individually passed in to
fetchEntry(TransactionContext, SyncOperation)
. Therefore,
it is important to make sure that the DatabaseChangeRecord instances
contain enough identifiable information (e.g. primary keys) for each entry
so that the entry can be found again.
The lifecycle of resync is similar to that of real-time sync, with a few differences:
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.entryType
- the type of database entry to be fetched (this is specified
on the CLI for the resync command)outputQueue
- a queue of DatabaseChangeRecord objects which will be individually
fetched via fetchEntry(TransactionContext, SyncOperation)
java.sql.SQLException
- if there is an error retrieving the list of entries to resync@Deprecated public void listAllEntries(TransactionContext ctx, java.util.Iterator<java.lang.String> inputLines, java.util.concurrent.BlockingQueue<DatabaseChangeRecord> outputQueue) throws java.sql.SQLException
entryType
parameter.
Gets a list of all the entries in the database from a given file input.
This is used by the 'resync' command line tool. The default implementation
throws a UnsupportedOperationException
; subclasses should override
if the resync functionality is needed for specific database records, which
can be specified in the input file.
The format for the inputLines
(e.g. the content of the file)
is user-defined; it may be key/value pairs, primary keys, or full SQL
statements, for example. The use of this method is triggered via the
--sourceInputFile argument on the resync CLI. The
outputQueue
should contain DatabaseChangeRecord
objects with the ChangeType
set to resync.
This method should not return until all the entries specified by the input
file have been added to the output queue. Separate threads will
concurrently drain entries from the queue and process them. The queue
should not actually contain full entries, but rather DatabaseChangeRecord
objects which identify the full database entries. These objects are then
individually passed in to
fetchEntry(TransactionContext, SyncOperation)
. Therefore,
it is important to make sure that the DatabaseChangeRecord instances
contain enough identifiable information (e.g. primary keys) for each entry
so that the entry can be found again.
The lifecycle of resync is similar to that of real-time sync, with a few differences:
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.inputLines
- an Iterator containing the lines from the specified input file to
resync (this is specified on the CLI for the resync command).
These lines can be any format, for example a set of primary keys,
a set of WHERE clauses, a set of full SQL queries, etc.outputQueue
- a queue of DatabaseChangeRecord objects which will be individually
fetched via fetchEntry(TransactionContext, SyncOperation)
java.sql.SQLException
- if there is an error retrieving the list of entries to resyncpublic void listAllEntries(TransactionContext ctx, java.lang.String entryType, java.util.Iterator<java.lang.String> inputLines, java.util.concurrent.BlockingQueue<DatabaseChangeRecord> outputQueue) throws java.sql.SQLException
UnsupportedOperationException
; subclasses should override
if the resync functionality is needed for specific database records, which
can be specified in the input file.
The format for the inputLines
(e.g. the content of the file)
is user-defined; it may be key/value pairs, primary keys, or full SQL
statements, for example. The use of this method is triggered via the
--sourceInputFile argument on the resync CLI. The
outputQueue
should contain DatabaseChangeRecord
objects with the ChangeType
set to resync.
This method should not return until all the entries specified by the input
file have been added to the output queue. Separate threads will
concurrently drain entries from the queue and process them. The queue
should not actually contain full entries, but rather DatabaseChangeRecord
objects which identify the full database entries. These objects are then
individually passed in to
fetchEntry(TransactionContext, SyncOperation)
. Therefore,
it is important to make sure that the DatabaseChangeRecord instances
contain enough identifiable information (e.g. primary keys) for each entry
so that the entry can be found again.
The lifecycle of resync is similar to that of real-time sync, with a few differences:
A TransactionContext
is provided, which allows
controlled access to the target database. The context will contain a fresh
fresh connection (i.e. a new transaction), and the Data Sync Server
will always commit or rollback the transaction automatically, depending on
whether this method returns normally or throws an exception. See the class
level documentation for warnings and additional details.
ctx
- a TransactionContext which provides a valid JDBC connection to the
database.entryType
- the type of database entry to be fetched (this is specified
on the CLI for the resync command)inputLines
- an Iterator containing the lines from the specified input file to
resync (this is specified on the CLI for the resync command).
These lines can be any format, for example a set of primary keys,
a set of WHERE clauses, a set of full SQL queries, etc.outputQueue
- a queue of DatabaseChangeRecord objects which will be individually
fetched via fetchEntry(TransactionContext, SyncOperation)
java.sql.SQLException
- if there is an error retrieving the list of entries to resync