#85 Improvements to job properties

Author: rjrudin

gradle.properties

@@ -1,3 +1,3 @@
 group=com.marklogic
 javadocsDir=../gh-pages-marklogic-java/javadocs
-version=3.2.0
+version=3.3.0-dev

src/main/java/com/marklogic/client/ext/datamovement/BatcherConfig.java

@@ -9,9 +9,12 @@
  */
 public class BatcherConfig extends LoggingObject {
 
+	public final static Integer DEFAULT_BATCH_SIZE = 100;
+	public final static Integer DEFAULT_THREAD_COUNT = 8;
+
 	private String jobName;
-	private Integer batchSize = 100;
-	private Integer threadCount = 8;
+	private Integer batchSize = DEFAULT_BATCH_SIZE;
+	private Integer threadCount = DEFAULT_THREAD_COUNT;
 	private ForestConfiguration forestConfig;
 
 	public void prepareBatcher(Batcher batcher) {

src/main/java/com/marklogic/client/ext/datamovement/QueryBatcherBuilder.java

@@ -9,5 +9,10 @@
  */
 public interface QueryBatcherBuilder {
 
+	/**
+	 * @param databaseClient      typically needed for constructing a QueryDefinition
+	 * @param dataMovementManager
+	 * @return
+	 */
 	QueryBatcher buildQueryBatcher(DatabaseClient databaseClient, DataMovementManager dataMovementManager);
 }

src/main/java/com/marklogic/client/ext/datamovement/job/AbstractQueryBatcherJob.java

@@ -3,20 +3,25 @@
 import com.marklogic.client.DatabaseClient;
 import com.marklogic.client.datamovement.*;
 import com.marklogic.client.ext.datamovement.*;
+import com.marklogic.client.ext.datamovement.listener.SimpleBatchLoggingListener;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Properties;
+import java.util.function.Consumer;
 
 /**
  * Provides basic plumbing for implementing QueryBatcherJob.
  */
-public abstract class AbstractQueryBatcherJob extends BatcherConfig implements QueryBatcherJob {
+public abstract class AbstractQueryBatcherJob extends BatcherConfig implements QueryBatcherJob, ConfigurableJob {
+
+	private List<JobProperty> jobProperties = new ArrayList<>();
 
 	private List<QueryBatchListener> urisReadyListeners;
 	private List<QueryFailureListener> queryFailureListeners;
 
-	private boolean applyConsistentSnapshot = false;
+	private boolean consistentSnapshot = false;
 	private boolean awaitCompletion = true;
 	private boolean stopJobAfterCompletion = true;
 
@@ -30,12 +35,17 @@ public abstract class AbstractQueryBatcherJob extends BatcherConfig implements Q
 	private String[] whereCollections;
 	private String whereUriPattern;
 	private String whereUrisQuery;
+	private boolean requireWhereProperty = true;
 
 	/**
 	 * @return a description of the job that is useful for logging purposes.
 	 */
 	protected abstract String getJobDescription();
 
+	protected AbstractQueryBatcherJob() {
+		addQueryBatcherJobProperties();
+	}
+
 	@Override
 	public QueryBatcherJobTicket run(DatabaseClient databaseClient) {
 		DataMovementManager dmm = this.dataMovementManager != null ? this.dataMovementManager : databaseClient.newDataMovementManager();
@@ -65,6 +75,77 @@ public QueryBatcherJobTicket run(DatabaseClient databaseClient) {
 		return new QueryBatcherJobTicket(dmm, queryBatcher, jobTicket);
 	}
 
+	@Override
+	public List<String> configureJob(Properties props) {
+		List<String> messages = new ArrayList<>();
+
+		for (JobProperty jobProperty : this.jobProperties) {
+			String name = jobProperty.getPropertyName();
+			String value = props.getProperty(name);
+			if (value != null && value.trim().length() > 0) {
+				jobProperty.getPropertyValueConsumer().accept(value);
+			} else if (jobProperty.isRequired()) {
+				messages.add("The property '" + name + "' is required");
+			}
+		}
+
+		if (requireWhereProperty && !isWherePropertySet() && queryBatcherBuilder == null) {
+			messages.add("At least one 'where' property must be set for selecting records to process");
+		}
+
+		return messages;
+	}
+
+	@Override
+	public List<JobProperty> getJobProperties() {
+		return jobProperties;
+	}
+
+	protected void addQueryBatcherJobProperties() {
+		addJobProperty("batchSize", "Number of records to process at once; defaults to " + DEFAULT_BATCH_SIZE,
+			value -> setBatchSize(Integer.parseInt(value)));
+
+		addJobProperty("consistentSnapshot", "Whether or not to apply a consistent snapshot to the query for records; defaults to false",
+			value -> setConsistentSnapshot(Boolean.parseBoolean(value)));
+
+		addJobProperty("jobName", "Optional name for the Data Movement job", value -> setJobName(value));
+
+		addJobProperty("logBatches", "Log each batch to stdout as it's processed",
+			value -> addUrisReadyListener(new SimpleBatchLoggingListener()));
+
+		addJobProperty("logBatchesWithLogger", "Log each batch as it's processed at the info-level using SLF4J",
+			value -> addUrisReadyListener(new SimpleBatchLoggingListener(true)));
+
+		addJobProperty("threadCount", "Number of threads to process records with; default to " + DEFAULT_THREAD_COUNT,
+			value -> setThreadCount(Integer.parseInt(value)));
+
+		addWhereJobProperties();
+	}
+
+	protected void addWhereJobProperties() {
+		addJobProperty("whereCollections", "Comma-delimited list of collections for selecting records to process",
+			value -> setWhereCollections(value.split(",")));
+
+		addJobProperty("whereUriPattern", "URI pattern for selecting records to process",
+			value -> setWhereUriPattern(value));
+
+		addJobProperty("whereUris", "Comma-delimited list of URIs for selecting records to process",
+			value -> setWhereUris(value.split(",")));
+
+		addJobProperty("whereUrisQuery", "CTS URIs query for selecting records to process",
+			value -> setWhereUrisQuery(value));
+	}
+
+	protected void addJobProperty(String name, String description, Consumer<String> propertyValueConsumer) {
+		jobProperties.add(new SimpleJobProperty(name, description, propertyValueConsumer));
+	}
+
+	protected void addRequiredJobProperty(String name, String description, Consumer<String> propertyValueConsumer) {
+		SimpleJobProperty prop = new SimpleJobProperty(name, description, propertyValueConsumer);
+		prop.setRequired(true);
+		jobProperties.add(prop);
+	}
+
 	/**
 	 * Can be overridden by the subclass to prepare the QueryBatcher before the job is started.
 	 *
@@ -73,7 +154,7 @@ public QueryBatcherJobTicket run(DatabaseClient databaseClient) {
 	protected void prepareQueryBatcher(QueryBatcher queryBatcher) {
 		super.prepareBatcher(queryBatcher);
 
-		if (applyConsistentSnapshot) {
+		if (consistentSnapshot) {
 			queryBatcher.withConsistentSnapshot();
 		}
 
@@ -91,30 +172,26 @@ protected void prepareQueryBatcher(QueryBatcher queryBatcher) {
 	}
 
 	/**
-	 * For subclasses to use to construct a QueryBatcherBuilder based on the "where" properties that have been set.
-	 *
 	 * @return
 	 */
 	protected QueryBatcherBuilder newQueryBatcherBuilder() {
 		if (queryBatcherBuilder != null) {
 			return queryBatcherBuilder;
 		}
 
-		if (isWherePropertySet()) {
-			if (whereUris != null && whereUris.length > 0) {
-				return new DocumentUrisQueryBatcherBuilder(whereUris);
-			}
-			if (whereCollections != null) {
-				return new CollectionsQueryBatcherBuilder(whereCollections);
-			}
-			if (whereUriPattern != null) {
-				return new UriPatternQueryBatcherBuilder(whereUriPattern);
-			}
-			if (whereUrisQuery != null) {
-				return new UrisQueryQueryBatcherBuilder(whereUrisQuery);
-			}
+		if (whereUris != null && whereUris.length > 0) {
+			return new DocumentUrisQueryBatcherBuilder(whereUris);
+		}
+		if (whereCollections != null) {
+			return new CollectionsQueryBatcherBuilder(whereCollections);
+		}
+		if (whereUriPattern != null) {
+			return new UriPatternQueryBatcherBuilder(whereUriPattern);
+		}
+		if (whereUrisQuery != null) {
+			return new UrisQueryQueryBatcherBuilder(whereUrisQuery);
 		}
-		throw new IllegalArgumentException("No 'where' property has been set, unable to construct a QueryBatcherBuilder");
+		return null;
 	}
 
 	/**
@@ -126,20 +203,20 @@ protected String getQueryDescription() {
 			return "with custom query";
 		}
 
-		if (isWherePropertySet()) {
-			if (whereUris != null && whereUris.length > 0) {
-				return "with URIs " + Arrays.asList(whereUris);
-			} else if (whereCollections != null && whereCollections.length > 0) {
-				return "in collections " + Arrays.asList(this.whereCollections);
-			} else if (whereUriPattern != null) {
-				return "matching URI pattern [" + whereUriPattern + "]";
-			} else if (whereUrisQuery != null) {
-				return "matching URIs query [" + whereUrisQuery + "]";
-			}
+		if (whereUris != null && whereUris.length > 0) {
+			return "with URIs " + Arrays.asList(whereUris);
+		} else if (whereCollections != null && whereCollections.length > 0) {
+			return "in collections " + Arrays.asList(this.whereCollections);
+		} else if (whereUriPattern != null) {
+			return "matching URI pattern [" + whereUriPattern + "]";
+		} else if (whereUrisQuery != null) {
+			return "matching URIs query [" + whereUrisQuery + "]";
 		}
-		throw new IllegalArgumentException("No 'where' property has been set, unable to construct a description of the query");
+
+		return null;
 	}
 
+
 	protected boolean isWherePropertySet() {
 		return
 			(whereUris != null && whereUris.length > 0)
@@ -217,12 +294,12 @@ public AbstractQueryBatcherJob setQueryFailureListeners(List<QueryFailureListene
 		return this;
 	}
 
-	public boolean isApplyConsistentSnapshot() {
-		return applyConsistentSnapshot;
+	public boolean isConsistentSnapshot() {
+		return consistentSnapshot;
 	}
 
-	public AbstractQueryBatcherJob setApplyConsistentSnapshot(boolean applyConsistentSnapshot) {
-		this.applyConsistentSnapshot = applyConsistentSnapshot;
+	public AbstractQueryBatcherJob setConsistentSnapshot(boolean consistentSnapshot) {
+		this.consistentSnapshot = consistentSnapshot;
 		return this;
 	}
 
@@ -253,4 +330,8 @@ public AbstractQueryBatcherJob setQueryBatcherBuilder(QueryBatcherBuilder queryB
 		this.queryBatcherBuilder = queryBatcherBuilder;
 		return this;
 	}
+
+	public void setRequireWhereProperty(boolean requireWhereProperty) {
+		this.requireWhereProperty = requireWhereProperty;
+	}
 }

src/main/java/com/marklogic/client/ext/datamovement/job/AddCollectionsJob.java

@@ -8,14 +8,24 @@ public class AddCollectionsJob extends AbstractQueryBatcherJob implements QueryB
 
 	private String[] collections;
 
+	public AddCollectionsJob() {
+		super();
+		addRequiredJobProperty("collections", "Comma-delimited list collections to which selected records are added",
+			value -> setCollections(value.split(",")));
+	}
+
 	public AddCollectionsJob(String... collections) {
-		this.collections = collections;
-		this.addUrisReadyListener(new AddCollectionsListener(collections));
+		this();
+		setCollections(collections);
 	}
 
 	@Override
 	protected String getJobDescription() {
 		return "Adding documents " + getQueryDescription() + " to collections " + Arrays.asList(collections);
 	}
 
+	public void setCollections(String... collections) {
+		this.collections = collections;
+		this.addUrisReadyListener(new AddCollectionsListener(collections));
+	}
 }

src/main/java/com/marklogic/client/ext/datamovement/job/AddPermissionsJob.java

@@ -8,14 +8,26 @@ public class AddPermissionsJob extends AbstractQueryBatcherJob {
 
 	private String[] rolesAndCapabilities;
 
+	public AddPermissionsJob() {
+		super();
+
+		addRequiredJobProperty("permissions",
+			"Comma-delimited list of roles and capabilities defining permissions added to selected records",
+			value -> setRolesAndCapabilities(value.split(",")));
+	}
+
 	public AddPermissionsJob(String... rolesAndCapabilities) {
-		this.rolesAndCapabilities = rolesAndCapabilities;
-		this.addUrisReadyListener(new AddPermissionsListener(rolesAndCapabilities));
+		this();
+		setRolesAndCapabilities(rolesAndCapabilities);
 	}
 
 	@Override
 	protected String getJobDescription() {
 		return "Adding permissions " + Arrays.asList(rolesAndCapabilities) + " to documents " + getQueryDescription();
 	}
 
+	public void setRolesAndCapabilities(String... rolesAndCapabilities) {
+		this.rolesAndCapabilities = rolesAndCapabilities;
+		this.addUrisReadyListener(new AddPermissionsListener(rolesAndCapabilities));
+	}
 }

src/main/java/com/marklogic/client/ext/datamovement/job/ConfigurableJob.java

@@ -0,0 +1,25 @@
+package com.marklogic.client.ext.datamovement.job;
+
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * Interface for a job to implement when it can be configured via a Properties object. A primary benefit for clients is
+ * that they can call the getJobProperties method and e.g. print out this list so a user knows what properties are
+ * available for a job.
+ */
+public interface ConfigurableJob {
+
+	/**
+	 * Configure this job with the given set of Properties.
+	 *
+	 * @return a list of strings, with each presumably being a validation error message
+	 */
+	List<String> configureJob(Properties props);
+
+	/**
+	 * @return the list of JobProperty objects for this job. One use case for this is for a client to print out the
+	 * name and description of each property.
+	 */
+	List<JobProperty> getJobProperties();
+}

src/main/java/com/marklogic/client/ext/datamovement/job/DeleteCollectionsJob.java

@@ -10,9 +10,15 @@ public class DeleteCollectionsJob extends AbstractQueryBatcherJob {
 
 	private String[] collections;
 
+	public DeleteCollectionsJob() {
+		super();
+		addRequiredJobProperty("collections", "Comma-delimited list of collections to delete",
+			value -> setCollections(value.split(",")));
+	}
+
 	public DeleteCollectionsJob(String... collections) {
-		this.collections = collections;
-		this.addUrisReadyListener(new DeleteListener());
+		this();
+		setCollections(collections);
 	}
 
 	@Override
@@ -24,4 +30,14 @@ protected QueryBatcherBuilder newQueryBatcherBuilder() {
 	protected String getJobDescription() {
 		return "Deleting collections: " + Arrays.asList(collections);
 	}
+
+	public void setCollections(String... collections) {
+		this.collections = collections;
+		this.addUrisReadyListener(new DeleteListener());
+	}
+
+	@Override
+	protected void addWhereJobProperties() {
+		// These don't apply to this job
+	}
 }