public interface WriteBatcher extends Batcher
To facilitate long-running write jobs, batches documents added by many external threads and coordinates internal threads to send the batches round-robin to all appropriate hosts in the cluster. Appropriate hosts are those containing a forest associated with the database for the DatabaseClient provided to DataMovementManager. Many external threads (threads not managed by WriteBatcher) can concurrently add documents by calling WriteBatcher add or addAs. Each time enough documents are added to make a batch, the batch is added to an internal queue where the first available internal thread will pick it up and write it to the server. Since batches are not written until they are full, you should always call flushAsync() or flushAndWait() when no more documents will be written to ensure that any partial batch is written.
Sample Usage:
WriteBatcher whb = dataMovementManager.newWriteBatcher()
.withBatchSize(100)
.withThreadCount(20)
.onBatchSuccess(batch -> {
logger.debug("batch # {}, so far: {}", batch.getJobBatchNumber(), batch.getJobResultsSoFar());
})
.onBatchFailure((batch,throwable) -> throwable.printStackTrace() );
JobTicket ticket = dataMovementManager.startJob(whb);
whb.add ("doc1.txt", new StringHandle("doc1 contents"));
whb.addAs("doc2.txt", "doc2 contents");
whb.flushAndWait(); // send the two docs even though they're not a full batch
dataMovementManager.stopJob(ticket);
Note: All Closeable content or metadata handles passed to add methods will be closed as soon as possible (after the batch is written). This is to avoid IO resource leakage. This differs from the normal usage of the Java Client API because WriteBatcher is asynchronous so there’s no easy way to know which handles have finished writing and can therefore be closed. So to save confusion we close all handles for you. If you have a resource that must be closed after a batch is written, but is not closed by your handle, override the close method of any Closeable handle and close your resource there.
| Modifier and Type | Method and Description |
|---|---|
WriteBatcher |
add(String uri,
AbstractWriteHandle contentHandle)
Add a document to be batched then written to the server when a batch is full or
flushAsync() or flushAndWait() is called. |
WriteBatcher |
add(String uri,
DocumentMetadataWriteHandle metadataHandle,
AbstractWriteHandle contentHandle)
Add a document to be batched then written to the server when a batch is full or
flushAsync() or flushAndWait() is called. |
WriteBatcher |
add(WriteEvent... docs)
Add docs in the form of WriteEvents.
|
WriteBatcher |
addAs(String uri,
DocumentMetadataWriteHandle metadataHandle,
Object content)
Add a document to be batched then written to the server when a batch is full or
flushAsync() or flushAndWait() is called. |
WriteBatcher |
addAs(String uri,
Object content)
Add a document to be batched then written to the server when a batch is full or
flushAsync() or flushAndWait() is called. |
boolean |
awaitCompletion()
Blocks until the job has finished or cancelled all queued tasks.
|
boolean |
awaitCompletion(long timeout,
TimeUnit unit)
Blocks until the job has finished or cancelled all queued tasks.
|
void |
flushAndWait()
Create a batch from any unbatched documents and write that batch, then wait for all batches to complete (the same as awaitCompletion().
|
void |
flushAsync()
Create a batch from any unbatched documents and write that batch asynchronously.
|
WriteFailureListener[] |
getBatchFailureListeners()
Get the array of WriteFailureListener instances registered via onBatchFailure including the HostAvailabilityListener registered by default.
|
WriteBatchListener[] |
getBatchSuccessListeners()
Get the array of WriteBatchListener instances registered via onBatchSuccess.
|
JobTicket |
getJobTicket()
After the job has been started, returns the JobTicket generated when the job was started.
|
String |
getTemporalCollection()
The temporal collection configured for temporal document inserts
|
ServerTransform |
getTransform() |
WriteBatcher |
onBatchFailure(WriteFailureListener listener)
Add a listener to run each time there is an exception writing a batch.
|
WriteBatcher |
onBatchSuccess(WriteBatchListener listener)
Add a listener to run each time a batch is successfully written.
|
void |
retry(WriteBatch queryEvent)
Retry in the same thread to send a batch that failed.
|
void |
setBatchFailureListeners(WriteFailureListener... listeners)
Remove any existing WriteFailureListener instances registered via onBatchFailure including the HostAvailabilityListener registered by default and replace them with the provided listeners.
|
void |
setBatchSuccessListeners(WriteBatchListener... listeners)
Remove any existing WriteBatchListener instances registered via onBatchSuccess and replace them with the provided listeners.
|
WriteBatcher |
withBatchSize(int batchSize)
Sets the number of documents to send per batch.
|
WriteBatcher |
withForestConfig(ForestConfiguration forestConfig)
If the server forest configuration changes mid-job, it can be re-fetched with
DataMovementManager.readForestConfig() then set via withForestConfig. |
WriteBatcher |
withJobName(String jobName)
Sets the job name.
|
WriteBatcher |
withTemporalCollection(String collection)
The temporal collection to use for a temporal document insert
|
WriteBatcher |
withThreadCount(int threadCount)
Sets the number of threads added to the internal thread pool for this instance to use for writing or reporting on batches of uris.
|
WriteBatcher |
withTransform(ServerTransform transform)
The ServerTransform to modify each document from each batch before it is written to the database.
|
getBatchSize, getForestConfig, getJobName, getThreadCount, isStoppedWriteBatcher add(String uri, AbstractWriteHandle contentHandle)
Add a document to be batched then written to the server when a batch is full or flushAsync() or flushAndWait() is called.
the Java Guide for more on using handles
uri - the document uricontentHandle - the document contentsWriteBatcher addAs(String uri, Object content)
Add a document to be batched then written to the server when a batch is full or flushAsync() or flushAndWait() is called.
IO Shortcut in MarkLogic Java Client API for more on using the *As shortcut methods
uri - the document uricontent - the document contentsWriteBatcher add(String uri, DocumentMetadataWriteHandle metadataHandle, AbstractWriteHandle contentHandle)
Add a document to be batched then written to the server when a batch is full or flushAsync() or flushAndWait() is called.
the Java Guide for more on using handles
uri - the document urimetadataHandle - the metadata (collection, permissions, metdata values, properties, quality)contentHandle - the document contentsWriteBatcher addAs(String uri, DocumentMetadataWriteHandle metadataHandle, Object content)
Add a document to be batched then written to the server when a batch is full or flushAsync() or flushAndWait() is called.
IO Shortcut in MarkLogic Java Client API for more on using the *As shortcut methods
uri - the document urimetadataHandle - the metadata (collection, permissions, metdata values, properties, quality)content - the document contentsWriteBatcher add(WriteEvent... docs)
Add docs in the form of WriteEvents. This is a convenience method for re-adding documents from failed batches.
docs - the batch of WriteEvents where each WriteEvent represents one documentWriteBatcher onBatchSuccess(WriteBatchListener listener)
Add a listener to run each time a batch is successfully written.
listener - the action which has to be done when the batch gets written successfullyWriteBatcher onBatchFailure(WriteFailureListener listener)
Add a listener to run each time there is an exception writing a batch.
These listeners will not run when an exception is thrown by a listener registered with onBatchSuccess. To learn more, please see Handling Exceptions in Listeners
listener - the code to run when a failure occursvoid retry(WriteBatch queryEvent)
Retry in the same thread to send a batch that failed. This method will throw an Exception if it fails again, so it can be wrapped in a try-catch block.
queryEvent - the information about the batch that failedWriteBatchListener[] getBatchSuccessListeners()
Get the array of WriteBatchListener instances registered via onBatchSuccess.
WriteFailureListener[] getBatchFailureListeners()
Get the array of WriteFailureListener instances registered via onBatchFailure including the HostAvailabilityListener registered by default.
void setBatchSuccessListeners(WriteBatchListener... listeners)
Remove any existing WriteBatchListener instances registered via onBatchSuccess and replace them with the provided listeners.
listeners - the WriteBatchListener instances this batcher should usevoid setBatchFailureListeners(WriteFailureListener... listeners)
Remove any existing WriteFailureListener instances registered via onBatchFailure including the HostAvailabilityListener registered by default and replace them with the provided listeners.
listeners - the WriteFailureListener instances this batcher should useWriteBatcher withTemporalCollection(String collection)
The temporal collection to use for a temporal document insert
collection - The temporal collection to use for a temporal document insertString getTemporalCollection()
The temporal collection configured for temporal document inserts
WriteBatcher withTransform(ServerTransform transform)
The ServerTransform to modify each document from each batch before it is written to the database.
transform - The ServerTransform to run on each document from each batch.ServerTransform getTransform()
WriteBatcher withForestConfig(ForestConfiguration forestConfig)
If the server forest configuration changes mid-job, it can be re-fetched with DataMovementManager.readForestConfig() then set via withForestConfig.
withForestConfig in interface BatcherforestConfig - the updated ForestConfigurationWriteBatcher withJobName(String jobName)
Sets the job name. Eventually, this may become useful for seeing named jobs in ops director.
withJobName in interface BatcherjobName - the name you would like to assign to this jobWriteBatcher withBatchSize(int batchSize)
Sets the number of documents to send per batch. Since documents are large relative to uris, this number should be much lower than the batch size for QueryBatcher. The default batch size is 100.
withBatchSize in interface BatcherbatchSize - the batch size – must be 1 or greaterWriteBatcher withThreadCount(int threadCount)
Sets the number of threads added to the internal thread pool for this instance to use for writing or reporting on batches of uris. Each time enough documents are added to fill a batch, a batch is created and a task is queued to write the batch. As a thread becomes available it grabs a task from the queue and performs the task (usually writing the batch to the server then reporting on the batch to listeners registered with onBatchSuccess and onBatchFailure). By default the number of threads is the number of hosts containing applicable forests. More threads should accommodate more throughput.
withThreadCount in interface Batchervoid flushAsync()
Create a batch from any unbatched documents and write that batch asynchronously.
void flushAndWait()
Create a batch from any unbatched documents and write that batch, then wait for all batches to complete (the same as awaitCompletion().
boolean awaitCompletion()
Blocks until the job has finished or cancelled all queued tasks.
boolean awaitCompletion(long timeout,
TimeUnit unit)
throws InterruptedException
Blocks until the job has finished or cancelled all queued tasks.
timeout - the maximum time to waitunit - the time unit of the timeout argumentInterruptedException - if interrupted while waitingJobTicket getJobTicket()
After the job has been started, returns the JobTicket generated when the job was started.
getJobTicket in interface BatcherIllegalStateException - if this job has not yet been startedCopyright © 2013-2017 MarkLogic Corporation.