Feature/import documents batch threads (#276)

* Defined new interface method and Javadoc for importDocuments that accepts two additional parameters batchSize and numThreads.

* Added dependency commons-collections4 so that we can use ListUtils.partition.

* Implemented new importDocuments method.

* Added test for importDocuments method with batchSize and numThreads.

* Minor refactoring of test for importDocuments method with batchSize and numThreads.

* added two false positive tests

Author: rkhaja

pom.xml

@@ -230,6 +230,10 @@
 			<artifactId>httpcore</artifactId>
 			<scope>provided</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.commons</groupId>
+			<artifactId>commons-collections4</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>commons-logging</groupId>
 			<artifactId>commons-logging</artifactId>
@@ -275,6 +279,11 @@
 				<artifactId>httpcore</artifactId>
 				<version>4.4.11</version>
 			</dependency>
+			<dependency>
+				<groupId>org.apache.commons</groupId>
+				<artifactId>commons-collections4</artifactId>
+				<version>4.4</version>
+			</dependency>
 			<dependency>
 				<groupId>commons-codec</groupId>
 				<artifactId>commons-codec</artifactId>

src/main/java/com/arangodb/ArangoCollection.java

@@ -150,6 +150,23 @@ <T> MultiDocumentEntity<DocumentCreateEntity<T>> insertDocuments(
 	 */
 	DocumentImportEntity importDocuments(Collection<?> values, DocumentImportOptions options) throws ArangoDBException;
 
+	/**
+	 * Bulk imports the given values into the collection.
+	 *
+	 * @param values
+	 *            a list of Objects that will be stored as documents
+	 * @param options
+	 *            Additional options, can be null
+	 * @param batchSize
+	 *            Size for individual data batches of the original collection of values
+	 * @param numThreads
+	 *            Number of parallel import threads
+	 * @return list of information about the imported batches
+	 * @throws ArangoDBException
+	 */
+	Collection<DocumentImportEntity> importDocuments(Collection<?> values, DocumentImportOptions options,
+													 int batchSize, int numThreads) throws ArangoDBException;
+
 	/**
 	 * Bulk imports the given values into the collection.
 	 * 

src/main/java/com/arangodb/internal/ArangoCollectionImpl.java

@@ -20,8 +20,15 @@
 
 package com.arangodb.internal;
 
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
+import org.apache.commons.collections4.ListUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -105,6 +112,32 @@ public DocumentImportEntity importDocuments(final Collection<?> values, final Do
 		return executor.execute(importDocumentsRequest(values, options), DocumentImportEntity.class);
 	}
 
+	@Override
+	public Collection<DocumentImportEntity> importDocuments(Collection<?> values, DocumentImportOptions options,
+															int batchSize, int numThreads) throws ArangoDBException {
+		List<? extends List<?>> batches = ListUtils.partition(new ArrayList<>(values), batchSize);
+		ExecutorService executorService = Executors.newFixedThreadPool(numThreads);
+		List<CompletableFuture<DocumentImportEntity>> completableFutureList = new ArrayList<>();
+		for (List<?> batch : batches) {
+			CompletableFuture<DocumentImportEntity> completableFuture = CompletableFuture.supplyAsync(() -> {
+				DocumentImportEntity documentImportEntity = importDocuments(batch, options);
+				return documentImportEntity;
+			}, executorService);
+			completableFutureList.add(completableFuture);
+		}
+		List<DocumentImportEntity> documentImportEntityList = new ArrayList<>();
+		for (CompletableFuture<DocumentImportEntity> completableFuture : completableFutureList) {
+			DocumentImportEntity documentImportEntity = null;
+			try {
+				documentImportEntity = completableFuture.get();
+			} catch (InterruptedException | ExecutionException e) {
+				throw new ArangoDBException(e);
+			}
+			documentImportEntityList.add(documentImportEntity);
+		}
+		return documentImportEntityList;
+	}
+
 	@Override
 	public DocumentImportEntity importDocuments(final String values) throws ArangoDBException {
 		return importDocuments(values, new DocumentImportOptions());

src/test/java/com/arangodb/ArangoCollectionTest.java

@@ -1879,6 +1879,64 @@ public void importDocumentsJsonFromToPrefix() {
 		}
 	}
 
+	@Test
+	public void importDocumentsBatchSizeNumThreads() {
+		final Collection<BaseDocument> values = new ArrayList<BaseDocument>();
+		for (int i = 1; i <= 100; i++) {
+			values.add(new BaseDocument(String.valueOf(i)));
+		}
+		int batchSize = 5;
+		int numThreads = 8;
+		final Collection<DocumentImportEntity> docsList = db.collection(COLLECTION_NAME).importDocuments(values,
+				new DocumentImportOptions(), batchSize, numThreads);
+		assertThat(docsList.size(), is(values.size() / batchSize));
+		for (final DocumentImportEntity docs : docsList) {
+			assertThat(docs, is(notNullValue()));
+			assertThat(docs.getCreated(), is(batchSize));
+			assertThat(docs.getEmpty(), is(0));
+			assertThat(docs.getErrors(), is(0));
+			assertThat(docs.getIgnored(), is(0));
+			assertThat(docs.getUpdated(), is(0));
+			assertThat(docs.getDetails(), is(empty()));
+		}
+	}
+
+	@Test
+	public void importDocumentsBatchSizeNumThreadsIllegalBatchSize() {
+		final Collection<BaseDocument> values = new ArrayList<BaseDocument>();
+		for (int i = 1; i <= 10; i++) {
+			values.add(new BaseDocument(String.valueOf(i)));
+		}
+
+		int batchSize = 0;
+		int numThreads = 8;
+
+		try {
+			final Collection<DocumentImportEntity> docsList = db.collection(COLLECTION_NAME).importDocuments(values,
+					new DocumentImportOptions(), batchSize, numThreads);
+			fail();
+		} catch (IllegalArgumentException e) {
+		}
+	}
+
+	@Test
+	public void importDocumentsBatchSizeNumThreadsIllegalNumThreads() {
+		final Collection<BaseDocument> values = new ArrayList<BaseDocument>();
+		for (int i = 1; i <= 10; i++) {
+			values.add(new BaseDocument(String.valueOf(i)));
+		}
+
+		int batchSize = 5;
+		int numThreads = 0;
+
+		try {
+			final Collection<DocumentImportEntity> docsList = db.collection(COLLECTION_NAME).importDocuments(values,
+					new DocumentImportOptions(), batchSize, numThreads);
+			fail();
+		} catch (IllegalArgumentException e) {
+		}
+	}
+
 	@Test
 	public void deleteDocumentsByKey() {
 		final Collection<BaseDocument> values = new ArrayList<BaseDocument>();